python同步库

⑥ Python 之测试环境db自动同步

多套测试环境，如何做基线的数据库级别的同步更新？

工作中测试环境有多套时，为保证基础环境配置的一致性，就需要所有测试环境的数据库结构保持一致。

例如：A需求在 beta1 环境进行测试，且A需求提测单中有新增表的 sql，B需求在 beta2 环境进行测试，由于A需求比B需求先发布上线，此时在B需求测试过程中发布时需要将主干的代码合并到当前需求分支(集成测试的需要，可以提前检测出已上线的需求是否对当前在测的需求有影响)，代码合并后对应的相关配置也得跟上，否则程序运行时会报错，所以就需要在 beta2 环境更新 beta1 环境A需求新增表的sql。

因为每一次的发布上线都会做数据库级别的同步更新，如果只是两、三个测试环境，使用人工来手动更新也是可以的，如果测试环境多且数据库更新的内容量大，依然使用人工手动更新，效率就会十分低下，同时也会造成一些人为操作的错误。这时自动化同步更新数据库就显得犹为重要了。在效率和正确率上都是完胜手工更新的。

由代码实现部分可以看出，有了这个自动同步的自动化脚本，在数据库更新时，只需要传入更新的 sql 语句就可一键自动同步多套测试环境的数据库信息了，十分高效。

⑦ 深入解析Python中的线程同步方法

深入解析Python中的线程同步方法
同步访问共享资源
在使用线程的时候，一个很重要的问题是要避免多个线程对同一变量或其它资源的访问冲突。一旦你稍不留神，重叠访问、在多个线程中修改（共享资源）等这些操作会导致各种各样的问题；更严重的是，这些问题一般只会在比较极端（比如高并发、生产服务器、甚至在性能更好的硬件设备上）的情况下才会出现。
比如有这样一个情况：需要追踪对一事件处理的次数
counter = 0

def process_item(item):
global counter
... do something with item ...
counter += 1
如果你在多个线程中同时调用这个函数，你会发现counter的值不是那么准确。在大多数情况下它是对的，但有时它会比实际的少几个。
出现这种情况的原因是，计数增加操作实际上分三步执行:
解释器获取counter的当前值计算新值将计算的新值回写counter变量
考虑一下这种情况：在当前线程获取到counter值后，另一个线程抢占到了CPU，然后同样也获取到了counter值，并进一步将counter值重新计算并完成回写；之后时间片重新轮到当前线程（这里仅作标识区分，并非实际当前），此时当前线程获取到counter值还是原来的，完成后续两步操作后counter的值实际只加上1。
另一种常见情况是访问不完整或不一致状态。这类情况主要发生在一个线程正在初始化或更新数据时，另一个进程却尝试读取正在更改的数据。
原子操作
实现对共享变量或其它资源的同步访问最简单的方法是依靠解释器的原子操作。原子操作是在一步完成执行的操作，在这一步中其它线程无法获得该共享资源。
通常情况下，这种同步方法只对那些只由单个核心数据类型组成的共享资源有效，譬如，字符串变量、数字、列表或者字典等。下面是几个线程安全的操作：
读或者替换一个实例属性读或者替换一个全局变量从列表中获取一项元素原位修改一个列表（例如：使用append增加一个列表项）从字典中获取一项元素原位修改一个字典（例如：增加一个字典项、调用clear方法）
注意，上面提到过，对一个变量或者属性进行读操作，然后修改它，最终将其回写不是线程安全的。因为另外一个线程会在这个线程读完却没有修改或回写完成之前更改这个共享变量/属性。
锁
锁是Python的threading模块提供的最基本的同步机制。在任一时刻，一个锁对象可能被一个线程获取，或者不被任何线程获取。如果一个线程尝试去获取一个已经被另一个线程获取到的锁对象，那么这个想要获取锁对象的线程只能暂时终止执行直到锁对象被另一个线程释放掉。
锁通常被用来实现对共享资源的同步访问。为每一个共享资源创建一个Lock对象，当你需要访问该资源时，调用acquire方法来获取锁对象（如果其它线程已经获得了该锁，则当前线程需等待其被释放），待资源访问完后，再调用release方法释放锁：
lock = Lock()

lock.acquire() #: will block if lock is already held
... access shared resource
lock.release()

注意，即使在访问共享资源的过程中出错了也应该释放锁，可以用try-finally来达到这一目的：
lock.acquire()
try:
... access shared resource
finally:
lock.release() #: release lock, no matter what

在Python 2.5及以后的版本中，你可以使用with语句。在使用锁的时候，with语句会在进入语句块之前自动的获取到该锁对象，然后在语句块执行完成后自动释放掉锁：
from __future__ import with_statement #: 2.5 only

with lock:
... access shared resource

acquire方法带一个可选的等待标识，它可用于设定当有其它线程占有锁时是否阻塞。如果你将其值设为False，那么acquire方法将不再阻塞，只是如果该锁被占有时它会返回False:
if not lock.acquire(False):
... 锁资源失败
else:
try:
... access shared resource
finally:
lock.release()

你可以使用locked方法来检查一个锁对象是否已被获取，注意不能用该方法来判断调用acquire方法时是否会阻塞，因为在locked方法调用完成到下一条语句（比如acquire）执行之间该锁有可能被其它线程占有。
if not lock.locked():
#: 其它线程可能在下一条语句执行之前占有了该锁
lock.acquire() #: 可能会阻塞

简单锁的缺点
标准的锁对象并不关心当前是哪个线程占有了该锁；如果该锁已经被占有了，那么任何其它尝试获取该锁的线程都会被阻塞，即使是占有锁的这个线程。考虑一下下面这个例子：
lock = threading.Lock()

def get_first_part():
lock.acquire()
try:
... 从共享对象中获取第一部分数据
finally:
lock.release()
return data

def get_second_part():
lock.acquire()
try:
... 从共享对象中获取第二部分数据
finally:
lock.release()
return data

示例中，我们有一个共享资源，有两个分别取这个共享资源第一部分和第二部分的函数。两个访问函数都使用了锁来确保在获取数据时没有其它线程修改对应的共享数据。
现在，如果我们想添加第三个函数来获取两个部分的数据，我们将会陷入泥潭。一个简单的方法是依次调用这两个函数，然后返回结合的结果：

def get_both_parts():
first = get_first_part()
seconde = get_second_part()
return first, second

这里的问题是，如有某个线程在两个函数调用之间修改了共享资源，那么我们最终会得到不一致的数据。最明显的解决方法是在这个函数中也使用lock:
def get_both_parts():
lock.acquire()
try:
first = get_first_part()
seconde = get_second_part()
finally:
lock.release()
return first, second

然而，这是不可行的。里面的两个访问函数将会阻塞，因为外层语句已经占有了该锁。为了解决这个问题，你可以通过使用标记在访问函数中让外层语句释放锁，但这样容易失去控制并导致出错。幸运的是，threading模块包含了一个更加实用的锁实现：re-entrant锁。
Re-Entrant Locks (RLock)

RLock类是简单锁的另一个版本，它的特点在于，同一个锁对象只有在被其它的线程占有时尝试获取才会发生阻塞；而简单锁在同一个线程中同时只能被占有一次。如果当前线程已经占有了某个RLock锁对象，那么当前线程仍能再次获取到该RLock锁对象。
lock = threading.Lock()
lock.acquire()
lock.acquire() #: 这里将会阻塞

lock = threading.RLock()
lock.acquire()
lock.acquire() #: 这里不会发生阻塞

RLock的主要作用是解决嵌套访问共享资源的问题，就像前面描述的示例。要想解决前面示例中的问题，我们只需要将Lock换为RLock对象，这样嵌套调用也会OK.
lock = threading.RLock()

def get_first_part():
... see above

def get_second_part():
... see above

def get_both_parts():
... see above

这样既可以单独访问两部分数据也可以一次访问两部分数据而不会被锁阻塞或者获得不一致的数据。
注意RLock会追踪递归层级，因此记得在acquire后进行release操作。
Semaphores

信号量是一个更高级的锁机制。信号量内部有一个计数器而不像锁对象内部有锁标识，而且只有当占用信号量的线程数超过信号量时线程才阻塞。这允许了多个线程可以同时访问相同的代码区。
semaphore = threading.BoundedSemaphore()
semaphore.acquire() #: counter减小

... 访问共享资源
semaphore.release() #: counter增大

当信号量被获取的时候，计数器减小；当信号量被释放的时候，计数器增大。当获取信号量的时候，如果计数器值为0，则该进程将阻塞。当某一信号量被释放，counter值增加为1时，被阻塞的线程（如果有的话）中会有一个得以继续运行。
信号量通常被用来限制对容量有限的资源的访问，比如一个网络连接或者数据库服务器。在这类场景中，只需要将计数器初始化为最大值，信号量的实现将为你完成剩下的事情。
max_connections = 10

semaphore = threading.BoundedSemaphore(max_connections)

如果你不传任何初始化参数，计数器的值会被初始化为1.
Python的threading模块提供了两种信号量实现。Semaphore类提供了一个无限大小的信号量，你可以调用release任意次来增大计数器的值。为了避免错误出现，最好使用BoundedSemaphore类，这样当你调用release的次数大于acquire次数时程序会出错提醒。
线程同步

锁可以用在线程间的同步上。threading模块包含了一些用于线程间同步的类。
Events

一个事件是一个简单的同步对象，事件表示为一个内部标识(internal flag)，线程等待这个标识被其它线程设定，或者自己设定、清除这个标识。
event = threading.Event()

#: 一个客户端线程等待flag被设定
event.wait()

#: 服务端线程设置或者清除flag
event.set()
event.clear()

一旦标识被设定，wait方法就不做任何处理（不会阻塞），当标识被清除时，wait将被阻塞直至其被重新设定。任意数量的线程可能会等待同一个事件。
Conditions

条件是事件对象的高级版本。条件表现为程序中的某种状态改变，线程可以等待给定条件或者条件发生的信号。
下面是一个简单的生产者/消费者实例。首先你需要创建一个条件对象：

#: 表示一个资源的附属项
condition = threading.Condition()
生产者线程在通知消费者线程有新生成资源之前需要获得条件：
#: 生产者线程
... 生产资源项
condition.acquire()
... 将资源项添加到资源中
condition.notify() #: 发出有可用资源的信号
condition.release()
消费者必须获取条件（以及相关联的锁），然后尝试从资源中获取资源项：
#: 消费者线程
condition.acquire()
while True:
...从资源中获取资源项
if item:
break
condition.wait() #: 休眠，直至有新的资源
condition.release()
... 处理资源

wait方法释放了锁，然后将当前线程阻塞，直到有其它线程调用了同一条件对象的notify或者notifyAll方法，然后又重新拿到锁。如果同时有多个线程在等待，那么notify方法只会唤醒其中的一个线程，而notifyAll则会唤醒全部线程。
为了避免在wait方法处阻塞，你可以传入一个超时参数，一个以秒为单位的浮点数。如果设置了超时参数，wait将会在指定时间返回，即使notify没被调用。一旦使用了超时，你必须检查资源来确定发生了什么。
注意，条件对象关联着一个锁，你必须在访问条件之前获取这个锁；同样的，你必须在完成对条件的访问时释放这个锁。在生产代码中，你应该使用try-finally或者with.
可以通过将锁对象作为条件构造函数的参数来让条件关联一个已经存在的锁，这可以实现多个条件公用一个资源：
lock = threading.RLock()
condition_1 = threading.Condition(lock)
condition_2 = threading.Condition(lock)

互斥锁同步
我们先来看一个例子：
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import time, threading

# 假定这是你的银行存款:
balance = 0
muxlock = threading.Lock()

def change_it(n):
# 先存后取，结果应该为0:
global balance
balance = balance + n
balance = balance - n

def run_thread(n):
# 循环次数一旦多起来，最后的数字就变成非0
for i in range(100000):
change_it(n)

t1 = threading.Thread(target=run_thread, args=(5,))
t2 = threading.Thread(target=run_thread, args=(8,))
t3 = threading.Thread(target=run_thread, args=(9,))
t1.start()
t2.start()
t3.start()
t1.join()
t2.join()
t3.join()
print balance

结果 :

[/data/web/test_python]$ python multhread_threading.py
0
[/data/web/test_python]$ python multhread_threading.py
61
[/data/web/test_python]$ python multhread_threading.py
0
[/data/web/test_python]$ python multhread_threading.py
24

上面的例子引出了多线程编程的最常见问题：数据共享。当多个线程都修改某一个共享数据的时候，需要进行同步控制。
线程同步能够保证多个线程安全访问竞争资源，最简单的同步机制是引入互斥锁。互斥锁为资源引入一个状态：锁定/非锁定。某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改；直到该线程释放资源，将资源的状态变成“非锁定”，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。

threading模块中定义了Lock类，可以方便的处理锁定：
#创建锁mutex = threading.Lock()
#锁定mutex.acquire([timeout])
#释放mutex.release()

其中，锁定方法acquire可以有一个超时时间的可选参数timeout。如果设定了timeout，则在超时后通过返回值可以判断是否得到了锁，从而可以进行一些其他的处理。
使用互斥锁实现上面的例子的代码如下：
balance = 0
muxlock = threading.Lock()

def change_it(n):
# 获取锁，确保只有一个线程操作这个数
muxlock.acquire()
global balance
balance = balance + n
balance = balance - n
# 释放锁，给其他被阻塞的线程继续操作
muxlock.release()

def run_thread(n):
for i in range(10000):
change_it(n)

加锁后的结果，就能确保数据正确：
[/data/web/test_python]$ python multhread_threading.py
0
[/data/web/test_python]$ python multhread_threading.py
0
[/data/web/test_python]$ python multhread_threading.py
0
[/data/web/test_python]$ python multhread_threading.py
0

⑧ 如何同步一个python字典有多重

介绍

似乎有很多的扶手椅建议和没有工作的例子。没有列在下面，甚至多重的答案，这是颇有几分失望和不安。由于python爱好者，我们应该支持我们的内置库，并同时并行处理和同步从来都不是小事，我相信它可以制成琐碎与适当的设计。这是很重要的现代多核架构，并且不能够underline！话虽如此

CodeGo.net，我远远满足多处理库，它仍然是处于起步阶段的阶段与不少缺陷，错误，并正在面向函数式编程（我讨厌）。目前我还是比较喜欢火焰兵模块（这是遥遥领先其对多重的因无法在服务器运行时共享新创建的对象多处理的严重限制。“注册”的管理对象只能实际注册一个对象之前管理（或服务器）开始数落够了，更多的代码：
Server.py
from multiprocessing.managers import SyncManager

class MyManager(SyncManager):
pass

syncdict = {}
def get_dict():
return syncdict
if __name__ == "__main__":
MyManager.register("syncdict", get_dict)
manager = MyManager(("127.0.0.1", 5000), authkey="password")
manager.start()
raw_input("Press any key to kill server".center(50, "-"))
manager.shutdown()

多重的SyncManager，可以提供同步共享对象的Server.py在上面的代码示例。此代码将不能工作在解释多处理库是如何找到“可赎回”为每个注册的对象相当棘手的运行。运行Server.py将启动一个定制SyncManager共享多个进程的syncdict字典，并且可以连接到客户端机或者机器上，或者，如果运行在比回环，其他机器的其他的IP地址。在这种情况下，服务器运行在环回（127.0.0.1）端口5000。操作syncdict当使用authkey使用安全连接。当任何一个键被按下的管理是关机。
Client.py
from multiprocessing.managers import SyncManager
import sys, time
class MyManager(SyncManager):
pass
MyManager.register("syncdict")
if __name__ == "__main__":
manager = MyManager(("127.0.0.1", 5000), authkey="password")
manager.connect()
syncdict = manager.syncdict()
print "dict = %s" % (dir(syncdict))
key = raw_input("Enter key to update: ")
inc = float(raw_input("Enter increment: "))
sleep = float(raw_input("Enter sleep time (sec): "))
try:
#if the key doesn't exist create it
if not syncdict.has_key(key):
syncdict.update([(key, 0)])
#increment key value every sleep seconds
#then print syncdict
while True:
syncdict.update([(key, syncdict.get(key) + inc)])
time.sleep(sleep)
print "%s" % (syncdict)
except KeyboardInterrupt:
print "Killed client"

客户端端还必须创建一个自定义的SyncManager，注册“syncdict”，这个没有传入一个可调用来检索共享字典。它定制到5000端口的环回IP地址（127.0.0.1）和authkey建立与管理的安全连接开始Server.py。它通过调用注册的调用上的管理器检索共享字典syncdict。它提示如下：
在syncdict的关键，操作上
该款项由键访问时的值每个周期
在几秒钟内每个周期sleep的金额
然后，客户端端会检查看是否存在的关键。如果它不它创建的syncdict关键。然后，客户端端进入一个“没完没了”循环，它会更新密钥的值由指定的金额sleep，并打印syncdict只是直到KeyboardInterrupt发生（CTRL
+C）重复此过程。
恼人的问题
该管理器启动，否则你会得到异常，即使在管理器中的目录调用就会发现，它确实是有被注册前管理人必须被调用。
该词典的所有的操作一定要做的，不是字典（syncdict [“爆破”]=2就会失败的路多处理器共享自定义对象）
使用SyncManager的将减轻恼人的问题＃2，除了恼人的问题＃1防止由SyncManager.dict（）被注册和共享返回的代理。
（SyncManager.dict（）只能被称为后的管理器启动，并注册只会工作的管理开始之前这样SyncManager.dict（）是做函数式编程和通过代理进程作为一个像文档的例子做的时候）
在服务器和客户端端必须注册，即使直觉它会看起来像客户端端将只能够连接到管理后，看着办吧（请添加到您的愿望清单多处理开发人员）
闭幕

我希望你喜欢这个相当彻底，稍有回答，就像我有。我有一个很大的trouble直在我的脑海，为什么我挣扎了这么多的多处理模块，其中火焰兵是一件轻而易举的，现在多亏了这个答案我已经击中了要害。我希望这给如何提高多处理器模块，因为我相信它有一个很大的承诺，但处于起步阶段达不到什么是可能的。尽管描述的恼人的问题，我认为这仍然是一个相当可行的替代方案，是非常简单的。你可以SyncManager.dict（），并把它传递给进程作为该文档显示方式，它很可能是一个更简单的解决方案，根据您的它只是感觉不自然
2.
有没有字典需要被摆在首位共享的理由吗？你可以有每一个线程维护自己的字典，并在线程处理或个别线程字典在一起的回拨副本年底实例？
我不知道到底你在做什么，所以请我的，我的书面计划可能无法正常工作一字不差。什么我的建议是更多的是一种高层次的设计理念。
3.
我将致力于一个单独的进程，以维护“共享字典”：例如： xmlrpclib使代码提供给其他进程少量，通过xmlrpclib如暴露一个函数，该函数key, increment执行和一个刚服用的key和返回值，用语义细节（是否有丢失的钥匙，等，等一个默认值），这取决于你的应用程序的需求。
那么你任何你喜欢的共享专用字典过程方法：从一个简单的字典一个单线程的服务器一直到一个简单的sqlite的数据库，等等，等等，我建议你开始与代码“就这么简单，你可以得到破除“（取决于你是否需要一个永久共享字典，或持久性是没有必要给你），和，如果需要优化。
4.
响应于适当的溶液中，以并行写入的问题。我做了非常快速的研究，发现这篇文章是在暗示一个锁定/信号的解决方案。 （
虽然这个例子是不是特异性的一本字典，我敢肯定，你可以编写一个基于类的包装对象，以帮助您基于这种想法词典的工作。
如果我有一个喜欢这个以线程安全的方式，我倒是Python的信号量解决方案。 （假设我的技术是行不通的。）我相信，信号量普遍放缓，由于其性质阻塞线程的效率。
从网站：
信号量是一个比较先进的信号量有一个内部的计数器，而不是一个锁标志，并且它只有块如果超过线程给定数量的试图保持信号量。根据如何在信号量被初始化，这允许多个线程访问代码段

semaphore = threading.BoundedSemaphore()
semaphore.acquire() # decrements the counter
... access the shared resource; work with dictionary, add item or whatever.
semaphore.release() # increments the counter