溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
?
?
?
線程同步：
線程間協同，通過某種技術，讓一個線程訪問某些數據時，其它線程不能訪問這些數據，直到該線程完成對數據的操作；
?
結果要能預知；
?
critical section，臨界區；
mutex，互斥量，threading.Lock類；?? #即鎖，你用我不用，我用你不用
semaphore，信號量；
event，事件，threading.Event類；
barrier，屏障、柵欄，threading.Barrier；
?
幫助：
Python 3.5.3 documentation-->Library Reference-->Concurrent Execution-->threading
?
?
?
目錄
threading.Event類：... 1
theading.Lock類：... 4
阻塞鎖：... 5
非阻塞鎖：... 10
?
?
?
event，事件，是線程間通信機制中最簡單的實現，使用一個內部的flag標記，通過flag的True或False的變化來進行操作；
?
e.set()，標記設置為True；
e.clear()，標記設置為False；
e.is_set()，標記是否為True；
e.wait(timeout=None)，Block until the internal flag is true，設置等待標記為True的時長，None為無限等待，等到返回True，未等到超時了返回False；
?
e.wait()的使用：
wait優于sleep，wait會主動讓出時間片，其它線程可以被調度，而sleep會占用時間片不讓出；
?
例：
def do(interval, e: threading.Event):
??? while not e.wait(interval):
??????? logging.info('do sth')
?
e = threading.Event()
threading.Thread(target=do, args=(3,e)).start()
e.wait(5)?? #可用time.sleep(5)，e.wait(5)優于time.sleep(5)
e.set()
print('main exit')
輸出：
2018-07-30-14:58:08?????? Thread info: 10268 Thread-1 do sth
main exit
?
例：
老板雇傭了一個工人，讓他生產杯子，老板一直等著工人，直到生產了10個杯子；
import threading
import time
import logging
FORMAT = '%(asctime)-15s\tThread info: %(thread)d %(threadName)s %(message)s'
logging.basicConfig(level=logging.INFO, format=FORMAT, datefmt='%Y-%m-%d-%H:%M:%S')
?
cups = []?? #當前，如果多個工人做則有問題
event = threading.Event()?? #實例，主線程、boss、worker都要看event，注意其作用域，使用同一個Event對象的flag
?
def boss(e: threading.Event):
??? logging.info("I'm boss,waiting for U.")
??? e.wait()
??? logging.info('good job.')
?
def worker(n, e: threading.Event):
??? while True:
??????? time.sleep(0.5)
??????? cups.append(1)
??????? logging.info('make 1')
??????? if len(cups) >= n:
??????????? # logging.info('I finished my job. {}'.format(len(cups)))?? #此處打印也可放到下面，老板在主線程中等待，if event.wait():中
??????????? e.set()
?? ?????????break
???????? logging.info('I finished my job. {}'.format(len(cups)))
?
b = threading.Thread(target=boss, args=(event,))
w = threading.Thread(target=worker, args=(10, event))
w.start()
b.start()
?
# if event.wait():?? #方1 ，老板在主線程中等待
# ????logging.info('I finished my job. {}'.format(len(cups)))
?
# while not event.wait(1):?? #方2，1秒看1次，老板在主線程中等待
#???? pass
# logging.info('I finished my job. {}'.format(len(cups)))
?
# while not event.is_set():?? #方3，老板在主線程中等待
#???? event.wait()?? #誰wait就是等到flag變為True，或等到超時返回False，不限制等待的個數
# logging.info('I finished my job. {}'.format(len(cups)))
?
輸出：
2018-07-30-14:47:12?????? Thread info: 11032 Thread-1 I'm boss,waiting for U.
2018-07-30-14:47:13?????? Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:13?????? Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:14?????? Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:14?????? Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:15?????? Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:15?????? Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:16?????? Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:16?????? Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:17?????? Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:17?????? Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:17?????? Thread info: 6996 Thread-2 I finished my job. 10
2018-07-30-14:47:17?????? Thread info: 11032 Thread-1 good job.
?
例：
Event練習：
實現Timer，延時執行的線程；
延時計算add(x,y)；
思路：
Timer的構造函數中參數得有哪些？
如何實現start啟動一個線程執行函數？
如何cancel取消待執行任務？
?
class Timer:
??? def __init__(self, interval, fn, *args, **kwargs):
??????? self.interval = interval
??????? self.fn = fn
??????? self.args = args
??????? self.kwargs = kwargs
??????? self.event = threading.Event()
?
??? def start(self):
??????? threading.Thread(target=self.__do).start()
?
??? def cancel(self):
??????? self.event.set()
?
??? def __do(self):
??????? self.event.wait(self.interval)?? #等待超時后返回False
?????????????????? # print(self.event.__dict__)?? # {'_flag': False, '_cond': <Condition(<unlocked _thread.lock object at 0x00000000012435D0>, 0)>}
??????? print(self.event.is_set())?? #False
??????? if not self.event.is_set():
??????????? self.fn(*self.args, **self.kwargs)
?
def add(x, y):
??? logging.info(x+y)
?
t = Timer(3, add, 4, 5)
t.start()
#t.cancel()?? #有此句self.is_set()為True
輸出：
False
2018-07-31-09:29:45?????? Thread info: 5156 Thread-1 9
?
?
?
解決多線程中數據同步；
?
凡是存在共享資源爭搶的地方都可使用鎖，從而保證只有一個使用者可完全使用這個資源；
?
程序=數據+算法，要保證數據的結果是可預知的，若不可預知該程序是沒用的，為得到正確的結果，性能是其次已不重要了；
?
一旦某一線程獲得鎖，其它試圖獲取的線程將被阻塞；
?
acquire(blocking=True,timeout=-1)，默認阻塞，阻塞可設置超時時間；非阻塞時，timeout禁止設置；成功獲取鎖，返回True，否則返回False；
release()，釋放鎖，可從任何線程調用釋放；該方法執行后，已上鎖的鎖會被重置為unlocked；在未上鎖的鎖上調用release()，拋RuntimeError: release unlocked lock；
?
注：
鎖將acquire()和release()中間的代碼管住；
大多數情況下用的是阻塞鎖；
鎖，類似ATM機的小房子；
加鎖，應在分析業務基礎之上，在合適的地方加，不能從前到后加一把大鎖，這樣效率低下；
?
加鎖、解鎖：
一般來說，加鎖后還要有一些代碼實現，在釋放鎖之前還有可能拋異常，一旦出現異常，鎖無法釋放，但當前線程可能因為這個異常終止了，這就產生了死鎖；
一般誰（某個線程）加的鎖誰解鎖；
常用語句：
try...finally，保證鎖的釋放；
with，Lock類中有__enter__()和__exit__()，鎖對象支持上下文管理；
?
鎖的應用場景：
鎖，適用于訪問和修改同一個共享資源的時候，即讀寫同一個資源的時候；
如果全部都是讀取同一個共享資源需要鎖嗎？不需要，因為這時可認為共享資源是不可變的，每一次讀取它都是一樣的值，所以不用加鎖；
不使用鎖，雖有了效率，但結果是錯的；
使用了鎖，雖效率低下，但結果是對的，所以為了對的結果，讓計算機計算去吧！
?
使用鎖的注意事項：
少用鎖，必要時加鎖，使用了鎖，多線程訪問被鎖的資源時，就成了串行，要么排隊執行，要么爭搶執行；
例如，高速公路上的車并行跑，到了省界只開放了一個收費口，過了這個口，車輛依然可在多車道上并行跑；過收費口時，如果排隊一輛一輛過，加不加鎖一樣，效率相當，但一旦出現爭搶，就必須加鎖一輛輛過；
加鎖時間越短越好，不需要就立即釋放鎖；
一定要避免死鎖，盡量用with語句；
?
?
lock.acquire()
?
例：
lock = threading.Lock()
lock.acquire()
print('get locker1')
lock.acquire()?? #block，再次獲取同一個鎖時阻塞（獨占），必須等別人將該鎖釋放才能獲取到，此行之后的語句不能執行；即，某一線程獲得鎖后，另一線程只要執行到lock.acquire()就會阻塞住（卡住）
print('get locker2')
lock.release()?? #一旦某一線程release，等著的線程爭搶，誰搶到誰上鎖；一般情況，有多少個acquire就有多少個release；要保證一定release，用到上下文（with語句）或try...finally
print('release locker')
?
注：
死鎖（解不開的鎖，就是死鎖）：
同一個鎖，你等我釋放、我等你釋放、自己等自己釋放；
兩個鎖，你等我放一把，我等你放一把，互相等待，都不釋放，都為阻塞狀態；
?
例：
def work():
??? time.sleep(3)
??? lock.release()?? #任何一個線程都可對同一個鎖操作，此處釋放鎖
?
lock = threading.Lock()
lock.acquire()
print('get locker1')
threading.Thread(target=work).start()
time.sleep(5)
lock.acquire()
print('get locker2')
lock.release()
print('release locker')
?
例：
需求：10個工人完成100個杯子；
注：臨界點問題：
當生產到99個時，若每個線程都可看到，每個線程看到轉頭去生產了，導致結果會多；
只允許一個線程可看到當前生產杯子的數量，其它線程不可見，這個線程看到后生產完1個杯子時，其它線程均可看到，杯子數量達到要求就不生產了；
?
cups = []
?
def worker(task=100):
??? while True:
??????? count = len(cups)
??????? logging.info('current count: {}'.format(count))
??????? if count >= task:
??????????? break
??????? cups.append(1)
??????? logging.info('{} make 1'.format(threading.current_thread()))
??? logging.info('total: {}'.format(count))
?
for _ in range(10):
??? threading.Thread(target=worker).start()
輸出：
……
2018-08-06-09:13:14?????? Thread info: 10208 Thread-2 current count: 104
2018-08-06-09:13:14?????? Thread info: 10208 Thread-2 total: 104
2018-08-06-09:13:14?????? Thread info: 10252 Thread-6 <Thread(Thread-6, started 10252)> make 1
2018-08-06-09:13:14?????? Thread info: 10252 Thread-6 current count: 104
2018-08-06-09:13:14?????? Thread info: 10252 Thread-6 total: 104
?
解決，加鎖：
cups = []
lock = threading.Lock()
?
def worker(lock:threading.Lock, task=100):
??? while True:
??????? lock.acquire()
??????? count = len(cups)
??????? logging.info('current count: {}'.format(count))
??????? lock.release()
??????? if count >= task:?? #雖不是大鎖，仍有問題，在臨界點時，其它線程仍可看到杯子總數，繼而再做；解決，把中間代碼全部放到鎖里
??????????? break
??????? lock.acquire()
??????? cups.append(1)
??????? lock.release()
??????? logging.info('{} make 1'.format(threading.current_thread()))
??? logging.info('total: {}'.format(count))
?
for _ in range(10):
??? threading.Thread(target=worker, args=(lock,)).start()
輸出：
……
2018-08-06-09:19:11?????? Thread info: 4092 Thread-4 current count: 100
2018-08-06-09:19:11?????? Thread info: 4092 Thread-4 total: 100
2018-08-06-09:19:11?????? Thread info: 11024 Thread-6 current count: 100
2018-08-06-09:19:11?????? Thread info: 11024 Thread-6 total: 100
?
例：
cups = []
lock = threading.Lock()
?
def worker(lock:threading.Lock, task=100):
??? while True:
??????? lock.acquire()?? #該鎖為互斥鎖，你有我沒有，我有你沒有
??????? count = len(cups)
??????? logging.info('current count: {}'.format(count))
??????? # lock.release()
??????? if count >= task:
??????????????????????????? lock.release()
???? ???????break?? #仍有問題，break后未釋放鎖，解決try...finally或上下文
??????? # lock.acquire()
??????? cups.append(1)
??????? lock.release()
??????? logging.info('{} make 1'.format(threading.current_thread()))
??? logging.info('total: {}'.format(count))
?
for _ in range(10):
??? threading.Thread(target=worker, args=(lock,)).start()
輸出：
……
2018-08-06-09:23:38?????? Thread info: 10148 Thread-5 current count: 99
2018-08-06-09:23:38?????? Thread info: 10148 Thread-5 <Thread(Thread-5, started 10148)> make 1
2018-08-06-09:23:38?????? Thread info: 10148 Thread-5 current count: 100
2018-08-06-09:23:38?????? Thread info: 10148 Thread-5 total: 100
?
例：
計數器類，可以加、可以減；
class Counter:
??? def __init__(self):
??????? self.__val = 0
?
??? def inc(self):
??????? self.__val += 1
?
??? def dec(self):
??????? self.__val -= 1
?
??? @property
??? def value(self):
??????? return self.__val
?
def do(c:Counter, count=100):
??? for _ in range(count):
??????? for i in range(-50,50):
??????????? if i < 0:
??????????????? c.dec()
??????????? else:
??????????????? c.inc()
?
c = Counter()
threadcount = 10?? #依次10,100,1000查看，結果沒變化
c2 = 1000?? #依次10,100,1000查看，當1000時，每個線程耗cpu時間長，結果將不可預知
for i in range(threadcount):
??? threading.Thread(target=do, args=(c,c2)).start()
# time.sleep(5)
# print(c.value)?? #多線程情況下，每個線程耗cpu時間長的情況下，此處打印的值不是最終結果
while True:
??? time.sleep(3)
??? print(threading.enumerate())
??? print(c.value)
輸出：
[<_MainThread(MainThread, started 7424)>]
81
[<_MainThread(MainThread, started 7424)>]
81
?
例，解決，加鎖：
class Counter:
??? def __init__(self):
??????? self.__val = 0
??????? self.__lock = threading.Lock()
?
??? def inc(self):
??????? try:?? #同with self.__lock: self.__val += 1;
??????????? self.__lock.acquire()
??????????? self.__val += 1
??????? finally:
??????????? self.__lock.release()
?
??? def dec(self):
??????? with self.__lock:
??????????? self.__val -= 1
?
??? @property
??? def value(self):
??????? with self.__lock:
??????????? return self.__val
?
def do(c:Counter, count=100):
??? for _ in range(count):
??????? for i in range(-50,50):
??????????? if i < 0:
??????????????? c.dec()
??????????? else:
??????????????? c.inc()
?
c = Counter()
threadcount = 10
c2 = 1000
for i in range(threadcount):
??? threading.Thread(target=do, args=(c,c2)).start()
# time.sleep(5)
# print(c.value)
while True:
??? time.sleep(1)
??? # print(threading.enumerate())
??? # print(c.value)
??? if threading.active_count() == 1:
??????? print(threading.enumerate())
??????? print(c.value)
??? else:
??????? print(threading.enumerate())
輸出：
[<Thread(Thread-2, started 11888)>, <Thread(Thread-3, started 8312)>, <Thread(Thread-6, started 10004)>, <_MainThread(MainThread, started 12200)>, <Thread(Thread-8, started 12012)>, <Thread(Thread-9, started 7788)>, <Thread(Thread-5, started 12032)>, <Thread(Thread-10, started 11576)>]
[<_MainThread(MainThread, started 12200)>]
0
[<_MainThread(MainThread, started 12200)>]
0
?
?
lock.acquire(False)，獲取到鎖返回True，否則返回False；
?
?
例：
lock = threading.Lock()
lock.acquire()
ret = lock.acquire(False)?? #獲取到鎖則返回True
print(ret)
輸出：
False
?
例：
cups = []
lock = threading.Lock()
?
def worker(lock:threading.Lock, task=100):
??? while True:
??????? if lock.acquire(False):
??????????? count = len(cups)
??????????? logging.info('current count: {}'.format(count))
??????????? if count >= task:
??????????????? lock.release()
??????????????? break
??????????? cups.append(1)
??????????? lock.release()
??????????? logging.info('{} make 1'.format(threading.current_thread()))
??? logging.info('total: {}'.format(count))
?
for _ in range(10):
??? threading.Thread(target=worker, args=(lock,)).start()
輸出：
……
2018-08-06-13:52:35?????? Thread info: 11232 Thread-9 current count: 100
2018-08-06-13:52:35?????? Thread info: 11232 Thread-9 total: 100
2018-08-06-13:52:35?????? Thread info: 11752 Thread-10 current count: 100
2018-08-06-13:52:35?????? Thread info: 11752 Thread-10 total: 100
?
例：
def worker(tasks):
??? for task in tasks:
??????? if task.lock.acquire(False):
??????????? time.sleep(0.01)
??????????? logging.info('{} {} begin to start'.format(threading.current_thread(), task.name))
??????? else:
??????????? logging.info('{} {} is working'.format(threading.current_thread(), task.name))
?
class Task:
??? def __init__(self, name):
??????? self.name = name
??????? self.lock = threading.Lock()
?
tasks = [Task('task-{}'.format(x)) for x in range(10)]
?
for i in range(5):
??? threading.Thread(target=worker, args=(tasks,), name='worker-{}'.format(i)).start()
輸出：
……
2018-08-06-14:10:34?????? Thread info: 12256 worker-1 <Thread(worker-1, started 12256)> task-7 is working
2018-08-06-14:10:34?????? Thread info: 12256 worker-1 <Thread(worker-1, started 12256)> task-8 is working
2018-08-06-14:10:34?????? Thread info: 12256 worker-1 <Thread(worker-1, started 12256)> task-9 is working
2018-08-06-14:10:34?????? Thread info: 11912 worker-3 <Thread(worker-3, started 11912)> task-9 begin to start
2018-08-06-14:10:34?????? Thread info: 10496 worker-4 <Thread(worker-4, started 10496)> task-8 begin to start
2018-08-06-14:10:34?????? Thread info: 10496 worker-4 <Thread(worker-4, started 10496)> task-9 is working
?
?
?
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
