Python多線程編程技術(shù)總結(jié)

學(xué)習(xí)并發(fā)編程

• 加速程序的運(yùn)行
• 高薪程序員必備能力

程序運(yùn)行的5種并發(fā)粒度

• 單線程
• 單線程多協(xié)程
• 多線程
• 多進(jìn)程
• 多機(jī)器

怎樣選擇并發(fā)技術(shù)

• 如果單機(jī)無法搞定

• 大數(shù)據(jù)計(jì)算

• IO密集型

• CPU經(jīng)常在等待IO
• 比如網(wǎng)絡(luò)爬蟲
• 選擇1：多協(xié)程 coroutine
• 選擇2：多線程 threading

• CPU密集型

• 計(jì)算密集型，CPU計(jì)算為主
• 比如加密解密
• 使用多進(jìn)程multithreading

線程池和進(jìn)程池

• 原理：提前創(chuàng)建好線程/進(jìn)程放在池子里，新的task到來可以重用這些資源，減少了新建、終止線程/進(jìn)程的開銷

• 池化的好處

• 提升性能：因?yàn)闇p去了大量新建、終止線程的開銷，重用了線程資源
• 適用場景：適合處理突發(fā)性大量請求或需要大量線程完成任務(wù)、但實(shí)際任務(wù)處理時(shí)間較短
• 防御功能：能有效避免系統(tǒng)因?yàn)閯?chuàng)建線程過多，而導(dǎo)致系統(tǒng)負(fù)荷過大響應(yīng)變慢等問題
• 代碼簡潔：使用線程池的語法比自己新建線程執(zhí)行線程更加簡潔

全局解釋器鎖GIL

任何時(shí)刻僅有一個(gè)線程在執(zhí)行。

在多核心處理器上，使用 GIL 的解釋器也只允許同一時(shí)間執(zhí)行一個(gè)線程

GIL目的：為了解決多線程之間數(shù)據(jù)完整性和狀態(tài)同步問題

GIL帶來的問題

• 即使使用了多線程，同一時(shí)刻也只有單個(gè)線程使用CPU，導(dǎo)致多核CPU的浪費(fèi)
• GIL只會(huì)對CPU密集型的程序產(chǎn)生影響
• 如果程序主要是在做I/O操作，比如處理網(wǎng)絡(luò)連接，那么多線程技術(shù)常常是一個(gè)明智的選擇

規(guī)避GIL的方法

• 規(guī)避方法2：使用multiprocessing多進(jìn)程，對CPU密集型計(jì)算，單獨(dú)啟動(dòng)子進(jìn)程解釋器去執(zhí)行
• 規(guī)避方法2：?將計(jì)算密集型的任務(wù)轉(zhuǎn)移到C語言中，因?yàn)镃語言比Python快得多，注意要在C語言中自己釋放GIL

多線程編程

應(yīng)用于IO密集型計(jì)算，比如幾乎所有的網(wǎng)絡(luò)后臺服務(wù)、網(wǎng)絡(luò)爬蟲

引入模塊

from?threading?import?Thread

新建、啟動(dòng)、等待結(jié)束

????t=Thread(target=func,?args=(100,?))
????t.start()
????t.join()

數(shù)據(jù)通信

import?queue
q?=?queue.Queue()
q.put(item)
item?=?q.get()

線程安全加鎖

from?threading?import?Lock
lock?=?Lock()
with?lock:
????#?do?something

信號量限制并發(fā)

from?threading?import?Semaphore
semaphre?=?Semaphore(10)
with?semaphre:
????#?do?something

使用線程池

from?concurrent.futures?import?ThreadPoolExecutor

with?ThreadPoolExecutor()?as?executor:
????#?方法1
????results?=?executor.map(func,?[1,2,3])

????#?方法2
????future?=?executor.submit(func,?1)
????result?=?future.result()

多進(jìn)程編程

應(yīng)用于CPU密集型計(jì)算，只有發(fā)現(xiàn)多線程編程有性能問題時(shí)，才求助于該模塊

引入模塊

from?multiprocessing?import?Process

新建、啟動(dòng)、等待結(jié)束

p?=?Process(target=f,?args=('bob',))
p.start()
p.join()

數(shù)據(jù)通信

from?multiprocessing?import?Queue
q?=?Queue()
q.put([42,?None,?'hello'])
item?=?q.get()

線程安全加鎖

from?multiprocessing?import?Lock
lock?=?Lock()
with?lock:
????#?do?something

信號量限制并發(fā)

from?multiprocessing?import?Semaphore
semaphore?=?Semaphore(10)
with?semaphore:
????#?do?something

進(jìn)程池

from?concurrent.futures?import?ProcessPoolExecutor

with?ProcessPoolExecutor()?as?executor:
????#?方法1
????results?=?executor.map(func,?[1,2,3])

????#?方法2
????future?=?executor.submit(func,?1)
????result?=?future.result()

多協(xié)程編程

異步編程的威力

• Nginx作為 Web 服務(wù)器：打敗了 同步阻塞服務(wù)器 Apache， 使用更少的資源支持更多的并發(fā)連接，體現(xiàn)更高的效率，能夠支持高達(dá) 50,000 個(gè)并發(fā)連接數(shù)的響應(yīng)，使用 epoll and kqueue 作為開發(fā)模型
• Redis為什么這么快：處理網(wǎng)絡(luò)請求采用單線程+使用多路I/O復(fù)用模型，非阻塞IO ，避免了不必要的上下文切換和競爭條件，也不存在多進(jìn)程或者多線程導(dǎo)致的切換而消耗 CPU，不用去考慮各種鎖的問題，不存在加鎖釋放鎖操作，沒有因?yàn)榭赡艹霈F(xiàn)死鎖而導(dǎo)致的性能消耗；
• Node.js的優(yōu)勢：采用事件驅(qū)動(dòng)、異步編程，為網(wǎng)絡(luò)服務(wù)而設(shè)計(jì)。其實(shí)Javascript的匿名函數(shù)和閉包特性非常適合事件驅(qū)動(dòng)、異步編程 Node.js非阻塞模式的IO處理帶來在相對低系統(tǒng)資源耗用下的高性能與出眾的負(fù)載能力，非常適合用作依賴其它IO資源的中間層服務(wù)
• Go語言的一個(gè)優(yōu)勢：Go 使用Goroutine 和 channel為生成協(xié)程和使用信道提供了輕量級的語法，使得編寫高并發(fā)的服務(wù)端軟件變得相當(dāng)容易，很多情況下完全不需要考慮鎖機(jī)制以及由此帶來的各種問題，相比Python單個(gè) Go 應(yīng)用也能有效的利用多個(gè) CPU 核，并行執(zhí)行的性能好

異步編程的原理

• 核心原理1：超級循環(huán)

• 在單線程內(nèi)實(shí)現(xiàn)并發(fā)
• 用一個(gè)超級循環(huán)（其實(shí)就是while true）循環(huán)，里面每次輪詢處理所有的task
• 記憶口訣：《the one loop》至尊循環(huán)馭眾生、至尊循環(huán)尋眾生、至尊循環(huán)引眾生、普照眾生欣欣榮

• 核心原理2：IO多路復(fù)用

• select

• poll

• epool

• 數(shù)據(jù)結(jié)構(gòu)：bitmap
• 最大連接數(shù)：1024
• fd拷貝：每次調(diào)用select拷貝
• 工作效率：輪詢O(N)
• 數(shù)據(jù)結(jié)構(gòu)：數(shù)組
• 最大連接數(shù)：無上限
• fd拷貝：每次調(diào)用poll拷貝
• 工作效率：輪詢O(N)
• 數(shù)據(jù)結(jié)構(gòu)：紅黑樹
• 最大連接數(shù)：無上限
• fd拷貝：fd首次調(diào)用epool_ctl拷貝，每次調(diào)用epoll_wait不拷貝
• 工作效率：回調(diào)O(1)
• 是一種同步IO模型，實(shí)現(xiàn)一個(gè)線程可以監(jiān)視多個(gè)文件句柄；
• 一旦某個(gè)文件句柄就緒，就能夠通知應(yīng)用程序進(jìn)行相應(yīng)的讀寫操作；
• 沒有文件句柄就緒時(shí)會(huì)阻塞應(yīng)用程序，交出cpu
• 多路是指網(wǎng)絡(luò)連接，復(fù)用指的是同一個(gè)線程

• 原理

• 3種實(shí)現(xiàn)方式

Python官方異步庫：asyncio

代碼例子

?import?asyncio

????#?獲取事件循環(huán)
????loop?=?asyncio.get_event_loop()

????#?定義協(xié)程
????async?def?myfunc(url):
????????await?get_url(url)

????#?創(chuàng)建task列表
????tasks?=?[loop.create_task(myfunc(url))?for?url?in?urls]

????#?執(zhí)行爬蟲事件列表
????loop.run_until_complete(asyncio.wait(tasks))

• 優(yōu)點(diǎn)：

• 官方庫支持
• 明確使用asyncio、await關(guān)鍵字編程，直觀易讀

• 缺點(diǎn)：

• 很多庫不支持，比如requests

Python第三方異步庫：Gevent

代碼例子

?import?gevent.monkey

????gevent.monkey.patch_all()

????import?gevent
????import?blog_spider
????import?time

????begin?=?time.time()
????for?url?in?blog_spider.urls:
????????blog_spider.craw(url)
????end?=?time.time()
????print("single?thread,?cost?=?",?end?-?begin)

????begin?=?time.time()
????tasks?=?[gevent.spawn(blog_spider.craw,?url)?for?url?in?blog_spider.urls]
????gevent.joinall(tasks)
????end?=?time.time()
????print("gevent,?cost?=?",?end?-?begin)

• 原理

• 提供猴子補(bǔ)丁MonkeyPatch方法，通過該方法gevent能夠 修改標(biāo)準(zhǔn)庫里面大部分的阻塞式系統(tǒng)調(diào)用，包括socket、ssl、threading和?select等模塊，而變?yōu)閰f(xié)作式運(yùn)行

• 優(yōu)點(diǎn)

• 只需要monkey.patch_all()，就能自動(dòng)修改阻塞為非阻塞
• 提供了pywsgi異步服務(wù)器可以封裝flask

• 缺點(diǎn)

• 不知道它具體patch了哪些庫修改了哪些模塊、類、函數(shù)
• 創(chuàng)造了“隱式的副作用”，如果出現(xiàn)問題很多時(shí)候極難調(diào)試

Gevent改造Flask為異步服務(wù)

代碼例子

from?gevent?import?monkey

monkey.patch_all()

from?flask?import?Flask
from?gevent?import?pywsgi

app?=?Flask(__name__)


@app.route("/")
def?index():
????return?"success"


if?__name__?==?"__main__":
????#?app.run()
????server?=?pywsgi.WSGIServer(("0.0.0.0",?8888),?app)
????server.serve_forever()