執行緒、進程、協程和GIL（三）

  • 2020 年 1 月 21 日
  • 筆記

上一篇文章介紹了：創建執行緒的兩種方式、Event對象判斷執行緒是否啟動、利用訊號量控制執行緒並發。

部落格鏈接：執行緒、進程、協程和GIL（二）

這一篇來說說執行緒間通訊的那些事兒： 

　　一個執行緒向另一個執行緒發送數據最安全的方式就是使用queue庫中的隊列了，通過創建一個供多個執行緒共享的Queue對象，這些執行緒使用put()和get()操作來向隊列中添加數據或者從隊列中取出數據，以達到執行緒間通訊的效果。

　　queue隊列基本方法：

　　　　queue.Queue(maxsize = num):  FIFO  先進先出隊列，如果maxsize小於或等於0 則代表隊列長度無線。

　　　　queue.LifoQueue(maxsize = num): LIFO 後進先出隊列(類似於棧)，如果maxsize小於或等於0 則代表隊列長度無線。

　　　　Queue.qsize(): 返回當前隊列中元素的個數

      　　 Queue.empty()   如果隊列為空，返回True,反之False 

      　　 Queue.full()   如果隊列滿了，返回True,反之False

      　　 Queue.get([block[, timeout]])   讀隊列，timeout等待時間 

      　　 Queue.put(item, [block[, timeout]])   寫隊列，timeout等待時間 

    　　   Queue.queue.clear()   清空隊列

　　使用Queue構造生產者消費者模型來實現執行緒間的通訊：

import time  from queue import Queue,LifoQueue  from threading import Thread    def producer(in_q):      while True:          time.sleep(1)          data = '包子'          if in_q.full() == True:              print('蒸籠滿了，放不下了')          in_q.put(data)  # 向隊列中塞東西          print('小明蒸%s！' %(data))    def customer(out_q):      while True:          time.sleep(3)          if out_q.empty() == True:                  print('小紅沒取到包子，餓死了！')          data = out_q.get() # 從隊列中取東西          print('小紅取 %s ' % (data))    if __name__ == '__main__':      q = Queue(maxsize=3)      t1 = Thread(target=producer, args=(q,))      t2 = Thread(target=customer, args=(q,))      t1.start()      t2.start()

　　上面的程式碼實現了一個簡單的生產者消費者，小明負責蒸包子，小紅負責吃包子。當隊列被包子塞滿時，小明就再也放不進去了，此時生產者這個執行緒就會阻塞。當小紅將隊列中的包子吃完時，消費者這個執行緒就會阻塞。因為Queue對象已經封裝了必要的鎖，所以我們可以在多個執行緒之間安全的功能共享數據。但是在生產者消費者的關閉問題會有一些麻煩，通用的解決方式就是在隊列中放置一個特殊值，當消費者讀到這個值時，就終止執行。

　　不過有個問題需要注意：向隊列中添加數據項時，並不會複製此數據項，執行緒間的通訊實際上是在執行緒間傳遞對象引用。如果你單線對象的共享狀態，那麼最好只傳遞不可修改的數據結構（如：整型、字元串、或者元組）或者一個對象的深拷貝。

給關鍵部分加鎖

執行緒的不安全：同一進程里執行緒是共享數據的，當各個執行緒訪問同一個數據資源時會出現競爭狀態，即數據可能會同時被多個執行緒佔用，造成數據混亂，這就是執行緒的不安全。

 　　為了保證執行緒安全，所以引進了互斥鎖，確保某段關鍵程式碼、共享數據只能由一個執行緒從頭到尾完整地執行：

　　顯式的加鎖：

from threading import Thread, Lock    num = 0  lock = Lock()  # 定義一個鎖    def run():      global num, lock  # 獲取全局變數      lock.acquire()  # 加鎖      num += 1      print(num)      lock.release()  # 釋放鎖    if __name__ == '__main__':      Thread_list = []      for i in range(1000):          t = Thread(target=run)          Thread_list.append(t)      for i in Thread_list:          i.start()

死鎖：但是加入互斥鎖之後有可能會產生一個問題：死鎖：若干子執行緒在系統資源競爭時，都在等待對方對某部分資源解除佔用狀態，結果誰也不願意先解鎖，互相等著，程式無法執行下去，這就是死鎖。

　　比如：有兩個執行緒一、二，兩個共享資源A、B，執行緒一給資源A加鎖，執行緒二給資源B加鎖，然後資源A需要訪問資源B，資源B需要調用資源A，執行緒一二雙方都在等待對方釋放鎖，所以就會造成死鎖。

　　　But、當程式設計師在加鎖之後忘記調用release()方法，或者加鎖之後程式拋異常導致不能正常釋放鎖，有可能會造成死鎖，為了避免這種情況，我們不需要顯式的手動加鎖和釋放鎖，而是使用with語句來進行自動控制:

from threading import Thread, Lock    num = 0  lock = Lock()  # 定義一個鎖    def run():      global num, lock      with lock:  # 自動的控制加鎖和釋放鎖          num += 1          print(num)    if __name__ == '__main__':      Thread_list = []      for i in range(1000):          t = Thread(target=run)          Thread_list.append(t)      for i in Thread_list:          i.start()

　　創建一個執行緒池：

　　concurrent.futures 函數庫有一個 ThreadPoolExecutor 類可以被用來完成這個任務

from concurrent.futures import ThreadPoolExecutor    def run():      print('我是子執行緒')    if __name__ == '__main__':      pool = ThreadPoolExecutor(max_workers=3)  # 創建一個容量為3的執行緒池      for i in range(3):          t = pool.submit(run,)  #在執行緒池中生成三個執行緒，他們都來調用run方法      print('我是主執行緒')

Tags:

VirMach 便宜 VPS