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