Python 執行緒池原理及實現
概述
傳統多執行緒方案會使用“即時建立， 即時銷燬”的策略。儘管與建立程序相比，建立執行緒的時間已經大大的縮短，但是如果提交給執行緒的任務是執行時間較短，而且執行次數極其頻繁，那麼伺服器將處於不停的建立執行緒，銷燬執行緒的狀態。
一個執行緒的執行時間可以分為3部分：執行緒的啟動時間、執行緒體的執行時間和執行緒的銷燬時間。在多執行緒處理的情景中，如果執行緒不能被重用，就意味著每次建立都需要經過啟動、銷燬和執行3個過程。這必然會增加系統相應的時間，降低了效率。
使用執行緒池：
由於執行緒預先被建立並放入執行緒池中，同時處理完當前任務之後並不銷燬而是被安排處理下一個任務，因此能夠避免多次建立執行緒，從而節省執行緒建立和銷燬的開銷，能帶來更好的效能和系統穩定性。
執行緒池原理圖.png
執行緒池模型
這裡使用建立Thread()例項來實現，下面會再用繼承threading.Thread()的類來實現
# 建立佇列例項， 用於儲存任務 queue = Queue() # 定義需要執行緒池執行的任務 def do_job(): while True: i = queue.get() time.sleep(1) print 'index %s, curent: %s' % (i, threading.current_thread()) queue.task_done() if __name__ == '__main__': # 建立包括3個執行緒的執行緒池 for i in range(3): t = Thread(target=do_job) t.daemon=True # 設定執行緒daemon 主執行緒退出，daemon執行緒也會推出，即時正在執行 t.start() # 模擬建立執行緒池3秒後塞進10個任務到佇列 time.sleep(3) for i in range(10): queue.put(i) queue.join()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
輸出結果
index 1, curent: <Thread(Thread-2, started daemon 139652180764416)> index 0, curent: <Thread(Thread-1, started daemon 139652189157120)> index 2, curent: <Thread(Thread-3, started daemon 139652172371712)> index 4, curent: <Thread(Thread-1, started daemon 139652189157120)> index 3, curent: <Thread(Thread-2, started daemon 139652180764416)> index 5, curent: <Thread(Thread-3, started daemon 139652172371712)> index 6, curent: <Thread(Thread-1, started daemon 139652189157120)> index 7, curent: <Thread(Thread-2, started daemon 139652180764416)> index 8, curent: <Thread(Thread-3, started daemon 139652172371712)> index 9, curent: <Thread(Thread-1, started daemon 139652189157120)> finish
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
可以看到所有任務都是在這幾個執行緒中完成Thread-(1-3)
執行緒池原理
執行緒池基本原理： 我們把任務放進佇列中去，然後開N個執行緒，每個執行緒都去佇列中取一個任務，執行完了之後告訴系統說我執行完了，然後接著去佇列中取下一個任務，直至佇列中所有任務取空，退出執行緒。
上面這個例子生成一個有3個執行緒的執行緒池，每個執行緒都無限迴圈阻塞讀取Queue佇列的任務所有任務都只會讓這3個預生成的執行緒來處理。
具體工作描述如下：
- 建立Queue.Queue()例項，然後對它填充資料或任務
- 生成守護執行緒池，把執行緒設定成了daemon守護執行緒
- 每個執行緒無限迴圈阻塞讀取queue佇列的專案item，並處理
- 每次完成一次工作後，使用queue.task_done()函式向任務已經完成的佇列傳送一個訊號
- 主執行緒設定queue.join()阻塞，直到任務佇列已經清空了，解除阻塞，向下執行
這個模式下有幾個注意的點：
將執行緒池的執行緒設定成daemon守護程序，意味著主執行緒退出時，守護執行緒也會自動退出，如果使用預設
daemon=False的話， 非daemon的執行緒會阻塞主執行緒的退出，所以即使queue佇列的任務已經完成
執行緒池依然阻塞無限迴圈等待任務，使得主執行緒也不會退出。當主執行緒使用了queue.join()的時候，說明主執行緒會阻塞直到queue已經是清空的，而主執行緒怎麼知道queue已經是清空的呢？就是通過每次執行緒queue.get()後並處理任務後，傳送queue.task_done()訊號，queue的資料就會減1，直到queue的資料是空的，queue.join()解除阻塞，向下執行。
這個模式主要是以佇列queue的任務來做主導的，做完任務就退出，由於執行緒池是daemon的，所以主退出執行緒池所有執行緒都會退出。 有別於我們平時可能以佇列主導thread.join()阻塞，這種執行緒完成之前阻塞主執行緒。看需求使用哪個join()：
如果是想做完一定數量任務的佇列就結束，使用queue.join()，比如爬取指定數量的網頁 如果是想執行緒做完任務就結束，使用thread.join()- 1
- 2
示例：使用執行緒池寫web伺服器
import socket
import threading
from threading import Thread
import threading
import sys
import time
import random
from Queue import Queue
host = ''
port = 8888
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind((host, port))
s.listen(3)
class ThreadPoolManger():
"""執行緒池管理器"""
def __init__(self, thread_num):
# 初始化引數
self.work_queue = Queue()
self.thread_num = thread_num
self.__init_threading_pool(self.thread_num)
def __init_threading_pool(self, thread_num):
# 初始化執行緒池，建立指定數量的執行緒池
for i in range(thread_num):
thread = ThreadManger(self.work_queue)
thread.start()
def add_job(self, func, *args):
# 將任務放入佇列，等待執行緒池阻塞讀取，引數是被執行的函式和函式的引數
self.work_queue.put((func, args))
class ThreadManger(Thread):
"""定義執行緒類，繼承threading.Thread"""
def __init__(self, work_queue):
Thread.__init__(self)
self.work_queue = work_queue
self.daemon = True
def run(self):
# 啟動執行緒
while True:
target, args = self.work_queue.get()
target(*args)
self.work_queue.task_done()
# 建立一個有4個執行緒的執行緒池
thread_pool = ThreadPoolManger(4)
# 處理http請求，這裡簡單返回200 hello world
def handle_request(conn_socket):
recv_data = conn_socket.recv(1024)
reply = 'HTTP/1.1 200 OK \r\n\r\n'
reply += 'hello world'
print 'thread %s is running ' % threading.current_thread().name
conn_socket.send(reply)
conn_socket.close()
# 迴圈等待接收客戶端請求
while True:
# 阻塞等待請求
conn_socket, addr = s.accept()
# 一旦有請求了，把socket扔到我們指定處理函式handle_request處理，等待執行緒池分配執行緒處理
thread_pool.add_job(handle_request, *(conn_socket, ))
s.close()- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
# 執行程序
[master][/data/web/advance_python/socket]$ python sock_s_threading_pool.py
# 檢視執行緒池狀況
[master][/data/web/advance_python/socket]$ ps -eLf|grep sock_s_threading_pool
lisa+ 27488 23705 27488 0 5 23:22 pts/30 00:00:00 python sock_s_threading_pool.py
lisa+ 27488 23705 27489 0 5 23:22 pts/30 00:00:00 python sock_s_threading_pool.py
lisa+ 27488 23705 27490 0 5 23:22 pts/30 00:00:00 python sock_s_threading_pool.py
lisa+ 27488 23705 27491 0 5 23:22 pts/30 00:00:00 python sock_s_threading_pool.py
lisa+ 27488 23705 27492 0 5 23:22 pts/30 00:00:00 python sock_s_threading_pool.py
# 跟我們預期一樣一共有5個執行緒，一個主執行緒，4個執行緒池執行緒- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
這個執行緒池web伺服器編寫框架包括下面幾個組成部分及步驟：
- 定義執行緒池管理器ThreadPoolManger，用於建立並管理執行緒池，提供add_job()介面，給執行緒池加任務
- 定義工作執行緒ThreadManger， 定義run()方法，負責無限迴圈工作佇列，並完成佇列任務
- 定義socket監聽請求s.accept() 和處理請求 handle_requests() 任務。
- 初始化一個4個執行緒的執行緒池，都阻塞等待這讀取佇列queue的任務
- 當socket.accept()有請求，則把conn_socket做為引數，handle_request方法，丟給執行緒池，等待執行緒池分配執行緒處理
GIL 對多執行緒的影響
因為Python的執行緒雖然是真正的執行緒，但直譯器執行程式碼時，有一個GIL鎖：Global Interpreter Lock，任何Python執行緒執行前，必須先獲得GIL鎖，然後，每執行100條位元組碼，直譯器就自動釋放GIL鎖，讓別的執行緒有機會執行。這個GIL全域性鎖實際上把所有執行緒的執行程式碼都給上了鎖，所以，多執行緒在Python中只能交替執行，即使100個執行緒跑在100核CPU上，也只能用到1個核。
但是對於IO密集型的任務，多執行緒還是起到很大效率提升，這是協同式多工
當一項任務比如網路 I/O啟動，而在長的或不確定的時間，沒有執行任何 Python 程式碼的需要，一個執行緒便會讓出GIL，從而其他執行緒可以獲取 GIL 而執行 Python。這種禮貌行為稱為協同式多工處理，它允許併發；多個執行緒同時等待不同事件。
兩個執行緒在同一時刻只能有一個執行 Python ，但一旦執行緒開始連線，它就會放棄 GIL ，這樣其他執行緒就可以執行。這意味著兩個執行緒可以併發等待套接字連線，這是一件好事。在同樣的時間內它們可以做更多的工作。
執行緒池要設定為多少？
伺服器CPU核數有限，能夠同時併發的執行緒數有限，並不是開得越多越好，以及執行緒切換是有開銷的，如果執行緒切換過於頻繁，反而會使效能降低
執行緒執行過程中，計算時間分為兩部分：
- CPU計算，佔用CPU
- 不需要CPU計算，不佔用CPU，等待IO返回，比如recv(), accept(), sleep()等操作，具體操作就是比如
訪問cache、RPC呼叫下游service、訪問DB，等需要網路呼叫的操作
那麼如果計算時間佔50%， 等待時間50%，那麼為了利用率達到最高，可以開2個執行緒：
假如工作時間是2秒， CPU計算完1秒後，執行緒等待IO的時候需要1秒，此時CPU空閒了，這時就可以切換到另外一個執行緒，讓CPU工作1秒後，執行緒等待IO需要1秒，此時CPU又可以切回去，第一個執行緒這時剛好完成了1秒的IO等待，可以讓CPU繼續工作，就這樣迴圈的在兩個執行緒之前切換操作。
那麼如果計算時間佔20%， 等待時間80%，那麼為了利用率達到最高，可以開5個執行緒：
可以想象成完成任務需要5秒，CPU佔用1秒，等待時間4秒，CPU線上程等待時，可以同時再啟用4個執行緒，這樣就把CPU和IO等待時間，最大化的重疊起來
抽象一下，計算執行緒數設定的公式就是：
N核伺服器，通過執行業務的單執行緒分析出本地計算時間為x，等待時間為y，則工作執行緒數（執行緒池執行緒數）設定為 N*(x+y)/x，能讓CPU的利用率最大化。
由於有GIL的影響，python只能使用到1個核，所以這裡設定N=1