python3-cookbook中每個小節以問題、解決方案和討論三個部分探討了Python3在某類問題中的最優解決方式，或者說是探討Python3本身的資料結構、函式、類等特性在某類問題上如何更好地使用，這本書對于加深Python3的理解和提升Python編程能力的都有顯著幫助，特別是對怎么提高Python程式的性能會有很好的幫助，如果有時間的話強烈建議看一下，
本文為學習筆記，文中的內容只是根據自己的作業需要和平時使用寫了書中的部分內容，并且文中的示例代碼大多直接貼的原文代碼，當然，代碼多數都在Python3.6的環境上都驗證過了的，不同領域的編程關注點也會有所不同，有興趣的可以去看全文，
python3-cookbook：https://python3-cookbook.readthedocs.io/zh_CN/latest/index.html
12.1 啟動與停止執行緒
Python中的執行緒除了使用is_alive()查詢它是否存活和使用join()將它加入到當前執行緒并等待它終止之外，并沒有提供多少可以對執行緒操作的方法，例如不能主動終止執行緒，不能給執行緒發送信號等，如果想要對執行緒進行別的查詢和操作，可以參考如下方案，
import time from threading import Thread class CountdownTask: def __init__(self): self._running = True def terminate(self): self._running = False def run(self, n): while self._running and n > 0: print('T-minus', n) n -= 1 time.sleep(5) c = CountdownTask() t = Thread(target=c.run, args=(10,)) t.start() # 主動終止執行緒 c.terminate() # 等待執行緒終止 t.join()
12.3 執行緒間通信
當你需要在執行緒間交換資料時，可以考慮使用queue庫中的佇列了，它的優勢在于其本身就是執行緒安全的，如果你使用的是其他的資料結構，就需要在代碼中手動添加執行緒鎖的相關操作了，需要注意的是，佇列的qsize()、full()和empty()等方法并不是執行緒安全的，例如當qsize()獲取結果為0時，可能另一個執行緒馬上就往佇列中添加了一個資料，此時qsize()的獲取結果就是1了，
對于佇列終止的判斷，可以通過在佇列中添加結束標志或者例外捕獲來判斷，當然，佇列的操作，還是要根據具體的場景來做，
from queue import Queue from threading import Thread # 佇列終止標志 _sentinel = object() def producer(out_q): while True: # 資料處理 ... # 向佇列中添加資料 out_q.put(data) out_q.put(_sentinel) def consumer(in_q): while True: # 從佇列獲取資料 data =https://www.cnblogs.com/guyuyun/p/ in_q.get() # 判斷佇列是否結束 if data is _sentinel: in_q.put(_sentinel) break # 資料處理 ... q = Queue() t1 = Thread(target=consumer, args=(q,)) t2 = Thread(target=producer, args=(q,)) t1.start() t2.start()
import queue q = queue.Queue() try: data = q.get(block=False) except queue.Empty: ... try: data = q.get(timeout=5.0) except queue.Empty: ... try: q.put(item, block=False) except queue.Full: ...
12.4 給關鍵部分加鎖
當你需要給可變物件添加鎖時，應該考慮使用with陳述句，而不是手動呼叫acquire方法和release方法，在進入with陳述句時會自動獲取鎖，離開with陳述句時則自動釋放鎖，
12.5 防止死鎖的加鎖機制
此小節主要記錄一個“哲學家就餐問題”的避免死鎖的解決方案，有興趣的可以看下，
哲學家就餐問題：五位哲學家圍坐在一張桌子前，每個人 面前有一個碗飯和一只筷子，在這里每個哲學家可以看做是一個獨立的執行緒，而每只筷子可以看做是一個鎖，每個哲學家可以處在靜坐、 思考、吃飯三種狀態中的一個，需要注意的是，每個哲學家吃飯是需要兩只筷子的，這樣問題就來了：如果每個哲學家都拿起自己左邊的筷子， 那么他們五個都只能拿著一只筷子坐在那兒，直到餓死，此時他們就進入了死鎖狀態，
import threading from contextlib import contextmanager # 執行緒運行時，local()回傳的實體會為每個執行緒創建一個屬于它自己的本地存盤，不同執行緒的本地存盤互不影響，且互不可見 _local = threading.local() # 利用背景關系管理器和鎖的id值進行排序來控制鎖的分配 @contextmanager def acquire(*locks): # Sort locks by object identifier locks = sorted(locks, key=lambda x: id(x)) # 每個執行緒第一次運行到這兒時，結果都是空串列 acquired = getattr(_local, 'acquired', []) if acquired and max(id(lock) for lock in acquired) >= id(locks[0]): raise RuntimeError('Lock Order Violation') # 為執行緒的本地存盤添加一個串列，存盤所有鎖的id值 acquired.extend(locks) _local.acquired = acquired try: for lock in locks: lock.acquire() yield finally: # Release locks in reverse order of acquisition for lock in reversed(locks): lock.release() del acquired[-len(locks):] # 5個哲學家就餐問題實作 # The philosopher thread def philosopher(left, right): while True: with acquire(left, right): print(threading.currentThread(), 'eating') # The chopsticks (represented by locks) NSTICKS = 5 chopsticks = [threading.Lock() for n in range(NSTICKS)] # Create all of the philosophers for n in range(NSTICKS): t = threading.Thread(target=philosopher, args=(chopsticks[n], chopsticks[(n + 1) % NSTICKS])) t.start()
12.6 保存執行緒的狀態資訊
threading.local()回傳的實體可以為每個執行緒創建一個本地存盤，即一個底層字典，不同執行緒之間的字典是不可見的，以下示例中，每個執行緒都有自己的專屬套接字連接，所以多執行緒運行時它們是互不影響的，
import threading from functools import partial from socket import socket, AF_INET, SOCK_STREAM class LazyConnection: def __init__(self, address, family=AF_INET, type=SOCK_STREAM): self.address = address self.family = AF_INET self.type = SOCK_STREAM self.local = threading.local() def __enter__(self): if hasattr(self.local, 'sock'): raise RuntimeError('Already connected') self.local.sock = socket(self.family, self.type) self.local.sock.connect(self.address) return self.local.sock def __exit__(self, exc_ty, exc_val, tb): self.local.sock.close() del self.local.sock def test(conn): with conn as s: s.send(b'GET /index.html HTTP/1.0\r\n') s.send(b'Host: www.python.org\r\n') s.send(b'\r\n') resp = b''.join(iter(partial(s.recv, 8192), b'')) print('Got {} bytes'.format(len(resp))) if __name__ == '__main__': conn = LazyConnection(('www.python.org', 80)) t1 = threading.Thread(target=test, args=(conn,)) t2 = threading.Thread(target=test, args=(conn,)) t1.start() t2.start() t1.join() t2.join()
12.7 創建一個執行緒池
如果程式中需要使用到執行緒池，或者需要執行緒所執行函式的回傳結果時，可以考慮使用from concurrent.futures import ThreadPoolExecutor，
import urllib.request from concurrent.futures import ThreadPoolExecutor def fetch_url(url): u = urllib.request.urlopen(url) data = u.read() return data # 創建執行緒池物件，并允許同時運行10個執行緒 pool = ThreadPoolExecutor(10) # 傳入執行緒要執行的函式，以及它的引數 a = pool.submit(fetch_url, 'http://www.python.org') b = pool.submit(fetch_url, 'http://www.pypy.org') # 獲取執行緒執行結果時，會阻塞當前執行緒，直到該執行緒執行完畢并回傳結果 x = a.result() y = b.result()
12.8 簡單的并行編程
如果想要進行CPU密集型運算，并且想利用CPU多核的特性，可以考慮使用concurrent.futures的ProcessPoolExecutor類，但是正如此小節標題所言，只能是執行一些簡單的函式形式，其他的類方法、閉包等形式并不支持，并且函式的引數和回傳結果也必須兼容pickle，
它的原理是創建N個獨立的Python解釋器來執行，N取決于系統CPU核心數，當然，在實體化時也可以指定ProcessPoolExecutor(N)，
可以使用對應map來批量執行函式，也可以使用submit來單獨執行某個函式，具體使用見示例，
# 使用map批量執行 from concurrent.futures import ProcessPoolExecutor def work(x): ... return result # 普通做法 # results = map(work, data) # 利用CPU多核特點 with ProcessPoolExecutor() as pool: results = pool.map(work, data)
from concurrent.futures import ProcessPoolExecutor def work(x): ... return result def when_done(r): print('Got:', r.result()) with ProcessPoolExecutor() as pool: ... # 單獨執行某個函式 future_result = pool.submit(work, arg) # 在使用result()獲取結果時，當前程式會被阻塞，直到產生結果 r = future_result.result() ... # 單獨執行如果不想被阻塞，可以使用add_done_callback指定一個回呼函式 # 這個函式接受一個Future實體引數，可以在回呼函式中獲取執行結果 future_result.add_done_callback(when_done)
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/187937.html
標籤：Python
下一篇：python 初學者