絕對(duì)干貨！Python 從業(yè)十年的程序員，寫(xiě)的萬(wàn)字經(jīng)驗(yàn)分享

本文起源于我在 Twitter 上發(fā)布的關(guān)于 Python 經(jīng)歷的一系列話(huà)題。

出于某些原因，想記錄一下我過(guò)去數(shù)年使用 Python 的經(jīng)驗(yàn)和一些感悟。畢竟算是一門(mén)把我?guī)牖ヂ?lián)網(wǎng)行業(yè)的語(yǔ)言，而我近期已經(jīng)幾乎不再寫(xiě) Py 代碼， 做一個(gè)記錄，也許會(huì)對(duì)他人起到些微的幫助，也算是紀(jì)念與感恩了。

最早接觸 py 是 2010 年左右，那之前主要是使用 c、fortran 和 matlab 做數(shù)值運(yùn)算。當(dāng)時(shí)在做一些文件文本處理時(shí)覺(jué)得很麻煩，后來(lái)看到 NASA 說(shuō)要用 py 取代 matlab，就去接觸了 py。

python 那極為簡(jiǎn)潔與優(yōu)美的語(yǔ)法給了當(dāng)時(shí)的我極大的震撼，時(shí)至今日，寫(xiě) py 代碼對(duì)我而言依然是一種帶有藝術(shù)意味的享受。

首先開(kāi)宗明義的說(shuō)一句：python 并不慢，至少不夠慢。拿一個(gè) web 后端來(lái)說(shuō)，一臺(tái)垃圾 4 核虛機(jī)，跑 4 個(gè)同步阻塞的 django，假設(shè) django 上合理利用線程分擔(dān)了阻塞操作，假設(shè)每節(jié)點(diǎn)每秒可以處理 50 個(gè)請(qǐng)求（超低估），在白天的 10 小時(shí)內(nèi)就可以處理 720 萬(wàn)請(qǐng)求。而這種機(jī)器跑一天僅需要 20 塊錢(qián)。

在學(xué)習(xí) Python 以前需要強(qiáng)調(diào)的是：基礎(chǔ)語(yǔ)法非常重要。雖然我們都不推崇過(guò)多的死記硬背，但是少量必要的死背是以后所有復(fù)雜思維活動(dòng)的基礎(chǔ)，就像五十音對(duì)于日語(yǔ)，通假字和常用動(dòng)名詞對(duì)于文言文，你不會(huì)就是不行。

一般認(rèn)為，這包括數(shù)據(jù)類(lèi)型（值/引用）、作用域（scope）、keyword、builtin 函數(shù)等

關(guān)于 Python 版本的選擇，很多公司老項(xiàng)目依然在用 2.6、2.7，新項(xiàng)目的話(huà)建議至少選擇 3.6（擁有穩(wěn)定的 asyncio）。

• 從 2.7 到 3.4 https://blog.laisky.com/p/whats-new-in-python3-4/
• 從 3.4 到 3.5 https://blog.laisky.com/p/whats-new-in-python3-5/
• 從 3.5 到 3.6 https://blog.laisky.com/p/whats-new-in-python3-6/
• 從 3.6 到 3.7 https://docs.python.org/zh-cn/3/whatsnew/3.7.html

關(guān)于版本最后在說(shuō)幾點(diǎn)，建議在本地和服務(wù)器上都通過(guò) pyenv 來(lái)管理版本，而不要去動(dòng)系統(tǒng)自帶的 python（以免引起額外的麻煩） https://blog.laisky.com/p/pyenv/

另外一點(diǎn)就是，如果你想寫(xiě)一個(gè)兼容 2、3 的工具包，你可以考慮使用 future http://python-future.org/compatible_idioms.html

最后提醒一下，2to3 這個(gè)腳本是有可能出錯(cuò)的。

學(xué)完基礎(chǔ)就可以開(kāi)始動(dòng)手寫(xiě)代碼了，這時(shí)候應(yīng)該謹(jǐn)記遵守一些“通行規(guī)范”，幾年前給公司內(nèi)分享時(shí)做過(guò)一個(gè)摘要：

• 風(fēng)格指引 https://laisky.github.io/style-guide-cn/style-guides/source-code-style-guides/
• 一些注意事項(xiàng) https://laisky.github.io/style-guide-cn/style-guides/consensuses/

有了一定的實(shí)踐經(jīng)驗(yàn)后，你應(yīng)該學(xué)習(xí)更多的包來(lái)提高自己的代碼水平。

• 值得學(xué)習(xí)的內(nèi)建包 https://pymotw.com/3/
• 值得了解的第三方包 https://github.com/vinta/awesome-python

因?yàn)?py 的哲學(xué)（import this）建議應(yīng)該有且僅有一個(gè)完美的方式做一件事，所以建議優(yōu)先采用且完善既有項(xiàng)目而不建議過(guò)多的造輪子。

一個(gè)小插曲，寫(xiě)這段的 Tim Peters 就是發(fā)明 timsort 的那位。

https://en.wikipedia.org/wiki/Tim_Peters_(software_engineer)

有空時(shí)候，建議盡可能的完整讀教材和文檔，建立系統(tǒng)性的知識(shí)體系，這可以極大的提升你的眼界和思維能力。我自己讀過(guò)且覺(jué)得值得推薦的針對(duì) py 的書(shū)籍有：

• https://docs.python.org/3/
• learning python
• 核心編程
• 改進(jìn)Python的91個(gè)建議
• Python高手之路
• Python源碼剖析
• 數(shù)據(jù)結(jié)構(gòu)與算法：Python語(yǔ)言描述

如果你真的很喜歡 Python 的話(huà)，那我覺(jué)得你應(yīng)該也會(huì)喜歡閱讀 PEP，記得幾年前我只要有空就會(huì)去翻閱 PEP，這相當(dāng)于是 Py 的 RFC，里面記錄了幾乎每一項(xiàng)語(yǔ)法的設(shè)計(jì)理念與目的。我特別喜歡的 PEP 有：

• 8
• 3148
• 380
• 484 & 3107
• 492: async
• 440
• 3132
• 495 你甚至能學(xué)到歷史知識(shí)

以前聽(tīng)別人講過(guò)一個(gè)比喻，靜態(tài)語(yǔ)言是吃冒菜，一次性燙好。而動(dòng)態(tài)語(yǔ)言是涮火鍋，吃一點(diǎn)涮一點(diǎn)。

那么我覺(jué)得，GIL 就是僅有一雙筷子的火鍋，即使你菜很多，一次也只能涮一個(gè)。

但是，對(duì)于 I/O bound 的操作，你不必一直夾著菜，而是可以?shī)A一些扔到鍋里，這樣就可以同時(shí)涮很多，提高并行效率。

GIL 在一個(gè)進(jìn)程內(nèi)，解釋器僅能同時(shí)解釋執(zhí)行一條語(yǔ)句，這為 py 提供了天然的語(yǔ)句級(jí)線程安全，從很多意義上說(shuō)，這都極大的簡(jiǎn)化了并行編程的難度。對(duì)于 I/O 型應(yīng)用，多線程并不會(huì)受到多大影響。對(duì)于 CPU 型應(yīng)用，編寫(xiě)一個(gè)基于 Queue 的多進(jìn)程 worker 其實(shí)也就是幾行的事。

(訂正：應(yīng)為偽指令級(jí)的線程安全)

from?time?import?sleep
from?concurrent.futures?import?ProcessPoolExecutor,?wait
from?multiprocessing?import?Manager,?Queue


N_PARALLEL?=?5


def?worker(i:?int,?q:?Queue)?->?None:
????print(f'worker?{i}?start')
????while?1:
????????data?=?q.get()
????????if?data?is?None:??#?采用毒丸（poison?pill）方式來(lái)結(jié)束進(jìn)程池
????????????q.put(data)
????????????print(f'worker?{i}?exit')
????????????return

????????print(f'dealing?with?data?{data}...')
????????sleep(1)




def?main():
????executor?=?ProcessPoolExecutor(max_workers=N_PARALLEL)??#?控制并發(fā)量
????with?Manager()?as?manager:
????????queue?=?manager.Queue(maxsize=50)??#?控制緩存量

????????workers?=?[executor.submit(worker,?i,?queue)?for?i?in?range(N_PARALLEL)]
????????for?i?in?range(50):
????????????queue.put(i)

????????print('all?task?data?submitted')

????????queue.put(None)
????????wait(workers)
????????print('all?done')


main()

我經(jīng)常給新人講，是否能謹(jǐn)慎的對(duì)待并行編程，是一個(gè)區(qū)分初級(jí)和資深后端開(kāi)發(fā)的分水嶺。業(yè)界有一句老話(huà)：“沒(méi)有正確的并行程序，只有不夠量的并行度”，由此可見(jiàn)并行開(kāi)發(fā)的復(fù)雜程度。

我個(gè)人認(rèn)為思考并行時(shí)主要是在考慮兩個(gè)問(wèn)題：同步控制和資源用量。

對(duì)于同步控制，你在 thread, multiprocessing, asyncio 幾個(gè)包里都會(huì)發(fā)現(xiàn)一系列的工具：

• Lock 互斥鎖
• RLock 可重入鎖
• Queue 隊(duì)列
• Condition 條件鎖
• Event 事件鎖
• Semaphore 信號(hào)量

這個(gè)就不展開(kāi)細(xì)談了，屬于另一個(gè)語(yǔ)言無(wú)關(guān)的大領(lǐng)域。（以前寫(xiě)過(guò)一個(gè)很簡(jiǎn)略的簡(jiǎn)介：并行編程中的各種鎖(https://blog.laisky.com/p/concurrency-lock/)）

對(duì)于資源控制，一般來(lái)說(shuō)主要就是兩個(gè)地方：

• 緩存區(qū)有多大（Queue 長(zhǎng)度）
• 并發(fā)量有多大（workers 數(shù)量）

一般來(lái)說(shuō)，前者直接確定了你內(nèi)存的消耗量，最好選擇一個(gè)恰好或略高于消費(fèi)量的數(shù)。后者一般直接決定了你的 CPU 使用率，過(guò)高的并發(fā)量會(huì)增加切換開(kāi)銷(xiāo)，得不償失。

既然提到了 workers，稍微簡(jiǎn)單展開(kāi)一下“池”這個(gè)概念。我們經(jīng)常提到線程池、進(jìn)程池、連接池。說(shuō)白了就是對(duì)于一些可重用的資源，不必每次都創(chuàng)建新的，而是使用完畢后回收留待下一個(gè)數(shù)據(jù)繼續(xù)使用。比如你可以選擇不斷地開(kāi)子線程，也可以選擇預(yù)先開(kāi)好一批線程，然后通過(guò) queue 來(lái)不斷的獲取和處理數(shù)據(jù)。

所以說(shuō)使用“池”的主要目的就是減少資源的消耗。另一個(gè)優(yōu)點(diǎn)是，使用池可以非常方便的控制并發(fā)度（很多新人以為 Queue 是用來(lái)控制并發(fā)度的，這是錯(cuò)誤的，Queue 控制的是緩存量）。

對(duì)于連接池，還有另一層好處，那就是端口資源是有限的，而且回收端口的速度很慢，你不斷的創(chuàng)建連接會(huì)導(dǎo)致端口迅速耗盡。

這里做一個(gè)用語(yǔ)的訂正。Queue 控制的應(yīng)該是緩沖量（buffer），而不是緩存量（cache）。一般來(lái)說(shuō)，我們習(xí)慣上將寫(xiě)入隊(duì)列稱(chēng)為緩沖，將讀取隊(duì)列稱(chēng)為緩存（有源）。

對(duì)前面介紹的 python 中進(jìn)程／線程做一個(gè)小結(jié)，線程池可以用來(lái)解決 I/O 的阻塞，而進(jìn)程可以用來(lái)解決 GIL 對(duì) CPU 的限制（因?yàn)槊恳粋€(gè)進(jìn)程內(nèi)都有一個(gè) GIL）。所以你可以開(kāi) N 個(gè)（小于等于核數(shù)）進(jìn)程池，然后在每一個(gè)進(jìn)程中啟動(dòng)一個(gè)線程池，所有的線程池都可以訂閱同一個(gè) Queue，來(lái)實(shí)現(xiàn)真正的多核并行。

非常簡(jiǎn)單的描述一下進(jìn)程／線程，對(duì)于操作系統(tǒng)而言，可以認(rèn)為進(jìn)程是資源的最小單位（在 PCB 內(nèi)保存如圖 1 的數(shù)據(jù)）。而線程是調(diào)度的最小單位。同一個(gè)進(jìn)程內(nèi)的線程共享除棧和寄存器外的所有數(shù)據(jù)。

所以在開(kāi)發(fā)時(shí)候，要小心進(jìn)程內(nèi)多線程數(shù)據(jù)的沖突，也要注意多進(jìn)程數(shù)據(jù)間的隔離（需要特別使用進(jìn)程間通信）

• 操作系統(tǒng)筆記：進(jìn)程(https://blog.laisky.com/p/os-process/)
• 操作系統(tǒng)筆記：調(diào)度(https://blog.laisky.com/p/os-scheduler/)

再簡(jiǎn)單的補(bǔ)充一下，進(jìn)程間通信的手段有：管道、信號(hào)、消息隊(duì)列、信號(hào)量、共享內(nèi)存和套接字。不過(guò)在 Py 里，單機(jī)上最常用的進(jìn)程間通信就是 multiprocessing 里的 Queue 和 sharedctypes。

順帶一提，因?yàn)?CPython 的 refcnt 機(jī)制，所以 COW（copy on write）并不可靠。

人們?cè)谝?jiàn)到別人的“錯(cuò)誤寫(xiě)法”時(shí)，傾向于無(wú)視或吐槽諷刺。但是這個(gè)行為除了讓自己爽一下外沒(méi)有任何意義，不懂的還是不懂，最后真正發(fā)揮影響的還是那些能夠描繪一整條學(xué)習(xí)路徑的方法。

我一直希望能看到一個(gè)“樸素誠(chéng)懇”的切合工程實(shí)踐的教程，而不是網(wǎng)上流傳的入門(mén)大全和網(wǎng)課兜售騙錢(qián)的框架調(diào)參速成。

關(guān)于進(jìn)程間的內(nèi)存隔離，補(bǔ)充一個(gè)簡(jiǎn)單直觀的例子?？梢钥吹狡胀ㄗ兞?normal_v在兩個(gè)子進(jìn)程內(nèi)變成了兩個(gè)獨(dú)立的變量（都輸出 1），而共享內(nèi)存的shared_v仍然是同一個(gè)變量，分別輸出了 1 和 2。

from?time?import?sleep
from?concurrent.futures?import?ProcessPoolExecutor,?wait
from?multiprocessing?import?Manager,?Queue
from?ctypes?import?c_int64


def?worker(i,?normal_v,?shared_v):
????normal_v?+=?1????????????#?因?yàn)檫M(jìn)程間內(nèi)存隔離，所以每個(gè)進(jìn)程都會(huì)得到?1
????shared_v.value?+=?1??????#?因?yàn)槭褂昧斯蚕韮?nèi)存，所以會(huì)分別得到?1?和?2

????print(f'worker[{i}]?got?normal_v?{normal_v},?shared_v?{shared_v.value}')


def?main():
????executor?=?ProcessPoolExecutor(max_workers=2)
????with?Manager()?as?manager:
????????lock?=?manager.Lock()
????????shared_v?=?manager.Value(c_int64,?0,?lock=lock)
????????normal_v?=?0

????????workers?=?[executor.submit(worker,?i,?normal_v,?shared_v)?for?i?in?range(2)]
????????wait(workers)
????????print('all?done')


main()

順帶一提，在 3.8 里有了 sharedmemory：

"""
shared?memory
=============

Output:
::
????worker[0]?got?normal_v?1,?shared_v?1
????worker[2]?got?normal_v?1,?shared_v?2
????worker[3]?got?normal_v?1,?shared_v?3
????worker[1]?got?normal_v?1,?shared_v?4
????worker[4]?got?normal_v?1,?shared_v?5
????worker[5]?got?normal_v?1,?shared_v?6
????worker[6]?got?normal_v?1,?shared_v?7
????worker[8]?got?normal_v?1,?shared_v?8
????worker[7]?got?normal_v?1,?shared_v?9
????worker[9]?got?normal_v?1,?shared_v?10
????all?done
"""

from?traceback?import?print_exc
from?time?import?sleep
from?concurrent.futures?import?ProcessPoolExecutor,?wait
from?multiprocessing?import?Event,?RLock
from?multiprocessing.shared_memory?import?ShareableList
from?multiprocessing.managers?import?SharedMemoryManager,?SyncManager
from?ctypes?import?c_int64


def?worker(l:?RLock,?evt:?Event,?i:?int,?normal_v:?int,?shared_v:?ShareableList):
????try:
????????evt.wait()??????#?確保任務(wù)同時(shí)開(kāi)始
????????normal_v?+=?1???#?因?yàn)檫M(jìn)程間內(nèi)存隔離，所以每個(gè)進(jìn)程都會(huì)得到?1
????????with?RLock():???#?需要自行處理鎖
????????????shared_v[0]?+=?1??#?因?yàn)槭褂昧斯蚕韮?nèi)存，所以會(huì)得到連續(xù)累加的值

????????print(f"worker[{i}]?got?normal_v?{normal_v},?shared_v?{shared_v[0]}")
????except?Exception:
????????print_exc()
????????raise


def?main():
????executor?=?ProcessPoolExecutor(max_workers=10)
????with?SharedMemoryManager()?as?smm,?SyncManager()?as?sm:
????????evt?=?sm.Event()
????????shared_v?=?smm.ShareableList([0])
????????normal_v?=?0
????????workers?=?[
????????????executor.submit(worker,?sm.RLock(),?evt,?i,?normal_v,?shared_v)
????????????for?i?in?range(10)
????????]

????????evt.set()
????????wait(workers)
????????[f.result()?for?f?in?workers]
????????print("all?done")


if?__name__?==?"__main__":
????main()

從過(guò)去的工作經(jīng)驗(yàn)中，我總結(jié)了一個(gè)簡(jiǎn)單粗暴的規(guī)矩：如果你要使用多進(jìn)程，那么在程序啟動(dòng)的時(shí)候就把進(jìn)程池啟動(dòng)起來(lái)，然后需要任何資源都請(qǐng)?jiān)谶M(jìn)程內(nèi)自行創(chuàng)建使用。如果有數(shù)據(jù)需要共享，一定要顯式的采用共享內(nèi)存或 queue 的方式進(jìn)行傳遞。

見(jiàn)過(guò)太多在進(jìn)程間共享不該共享的東西而導(dǎo)致的極為詭異的數(shù)據(jù)行為。

最早，一臺(tái)機(jī)器從頭到尾只能干一件事情。

后來(lái)，有了分時(shí)系統(tǒng)，我們可以開(kāi)很多進(jìn)程，同時(shí)干很多事。

但是進(jìn)程的上下文切換開(kāi)銷(xiāo)太大，所以又有了線程，這樣一個(gè)核可以一直跑一個(gè)進(jìn)程，而僅需要切換進(jìn)程內(nèi)子線程的棧和寄存器。

直到遇到了 C10K 問(wèn)題，人們發(fā)覺(jué)切換幾萬(wàn)個(gè)線程還是挺重的，是否能更輕？

這里簡(jiǎn)單的展開(kāi)一下，內(nèi)存在操作系統(tǒng)中會(huì)被劃分為內(nèi)核態(tài)和用戶(hù)態(tài)兩部分，內(nèi)核態(tài)供內(nèi)核運(yùn)行，用戶(hù)態(tài)供普通的程序用。

應(yīng)用程序通過(guò)系統(tǒng) API（俗稱(chēng) syscall）和內(nèi)核發(fā)生交互。拿常見(jiàn)的 HTTP 請(qǐng)求來(lái)說(shuō)，其實(shí)就是一次同步阻塞的 socket 調(diào)用，每次調(diào)用都會(huì)導(dǎo)致線程阻塞等待內(nèi)核響應(yīng)（內(nèi)核陷入）。

而被阻塞的線程就會(huì)導(dǎo)致切換的發(fā)生。所以自然會(huì)問(wèn)，能不能減少這種切換開(kāi)銷(xiāo)？換句話(huà)說(shuō)，能不能在一個(gè)地方把事情做完，而不要切來(lái)切去的。

這個(gè)問(wèn)題有兩個(gè)解決思路，一是把所有的工作放進(jìn)內(nèi)核去做（略）。

另一個(gè)思路就是把盡可能多的工作放到用戶(hù)態(tài)來(lái)做。這需要內(nèi)核接口提供額外的支持：異步系統(tǒng)調(diào)用。

如 socket 這樣的調(diào)用就支持非阻塞調(diào)用，調(diào)用后會(huì)拿到一個(gè)未就緒的 fp，將這個(gè) fp 交給負(fù)責(zé)管理 I/O 多路復(fù)用的 selector，再注冊(cè)好需要監(jiān)聽(tīng)的事件和回調(diào)函數(shù)（或者像 tornado 一樣采用定時(shí) poll），就可以在事件就緒（如 HTTP 請(qǐng)求的返回已就緒）時(shí)執(zhí)行相關(guān)函數(shù)。

https://github.com/tornadoweb/tornado/blob/f1824029db933d822f5b0d02583e4e6137f2bfd2/tornado/ioloop.py#L746

這樣就可以實(shí)現(xiàn)在一個(gè)線程內(nèi)，啟動(dòng)多個(gè)曾經(jīng)會(huì)導(dǎo)致線程被切換的系統(tǒng)調(diào)用，然后在一個(gè)線程內(nèi)監(jiān)聽(tīng)這些調(diào)用的事件，誰(shuí)先就緒就處理誰(shuí)，將切換的開(kāi)銷(xiāo)降到了最小。

有一個(gè)需要特別注意的要點(diǎn)，你會(huì)發(fā)現(xiàn)主線程其實(shí)就是一個(gè)死循環(huán)，所有的調(diào)用都發(fā)生在這個(gè)循環(huán)之內(nèi)。所以，你寫(xiě)的代碼一定要避免任何阻塞。

聽(tīng)上去很美好，這是個(gè)萬(wàn)能方案嗎？

很可惜不是的，最直接的一個(gè)問(wèn)題是，并不是所有的 syscall 都提供了異步方法，對(duì)于這種調(diào)用，可以用線程池進(jìn)行封裝。對(duì)于 CPU 密集型調(diào)用，可以用進(jìn)程池進(jìn)行封裝，asyncio 里提供了 executor 和協(xié)程進(jìn)行聯(lián)動(dòng)的方法，這里提供一個(gè)線程池的簡(jiǎn)單例子，進(jìn)程池其實(shí)同理。

from?time?import?sleep
from?asyncio?import?get_event_loop,?sleep?as?asleep,?gather,?ensure_future
from?concurrent.futures?import?ThreadPoolExecutor,?wait,?Future
from?functools?import?wraps


executor?=?ThreadPoolExecutor(max_workers=10)
ioloop?=?get_event_loop()


def?nonblocking(func)?->?Future:
????@wraps(func)
????def?wrapper(*args):
????????return?ioloop.run_in_executor(executor,?func,?*args)
????return?wrapper


@nonblocking??#?用線程池封裝沒(méi)法協(xié)程化的普通阻塞程序
def?foo(n:?int):
????"""假裝我是個(gè)很耗時(shí)的阻塞調(diào)用"""
????print('start?blocking?task...')
????sleep(n)
????print('end?blocking?task')


async?def?coroutine_demo(n:?int):
????"""我就是個(gè)普通的協(xié)程"""

????#?協(xié)程內(nèi)不能出現(xiàn)任何的阻塞調(diào)用，所謂一朝協(xié)程，永世協(xié)程
????#?那我偏要調(diào)一個(gè)普通的阻塞函數(shù)怎么辦？
????#?最簡(jiǎn)單的辦法，套一個(gè)線程池…
????await?foo(n)


async?def?coroutine_demo_2():
????print('start?coroutine?task...')
????await?asleep(1)
????print('end?coroutine?task')


async?def?coroutine_main():
????"""一般我們會(huì)寫(xiě)一個(gè)?coroutine?的?main?函數(shù)，專(zhuān)門(mén)負(fù)責(zé)管理協(xié)程"""
????await?gather(
????????coroutine_demo(1),
????????coroutine_demo_2()
????)


def?main():
????ioloop.run_until_complete(coroutine_main())
????print('all?done')


main()

• Python3 asyncio 簡(jiǎn)介(https://blog.laisky.com/p/asyncio/)

上面的例子全部都基于 3.7，如果你還在使用 Py2，那么你也可以通過(guò) gevent、tornado 用上協(xié)程。

我個(gè)人傾向于 tornado，因?yàn)楦鼮榘缀校覍?xiě)法和 3 接近，如果你也贊同，那么可以試試我以前給公司寫(xiě)的 kipp 庫(kù)，基于 tornado 封裝了更多的工具。

https://github.com/Laisky/kipp/blob/2bc5bda6e7f593f89be662f46fed350c9daabded/kipp/aio/init.py

Gevent Demo:

#!/usr/bin/env?python
#?-*-?coding:?utf-8?-*-

"""
Gevent?Pool?&?Child?Tasks
=========================

You?can?use?gevent.pool.Pool?to?limit?the?concurrency?of?coroutines.

And?you?can?create?unlimit?subtasks?in?each?coroutine.


Benchmark
=========

????cost?2.675039052963257s?for?url?http://httpbin.org/
????cost?2.66813588142395s?for?url?http://httpbin.org/ip
????cost?2.674264907836914s?for?url?http://httpbin.org/user-agent
????cost?2.6776888370513916s?for?url?http://httpbin.org/get
????cost?3.97711181640625s?for?url?http://httpbin.org/headers
????total?cost?3.9886841773986816s
"""
import?time

import?gevent
from?gevent.pool?import?Pool
import?gevent.monkey


pool?=?Pool(10)??#?set?the?concurrency?limit
gevent.monkey.patch_socket()

try:
????import?urllib2
except?ImportError:
????import?urllib.request?as?urllib2


TARGET_URLS?=?(
????'http://httpbin.org/',
????'http://httpbin.org/ip',
????'http://httpbin.org/user-agent',
????'http://httpbin.org/headers',
????'http://httpbin.org/get',
)


def?demo_child_task():
????"""Sub?coroutine?task"""
????gevent.sleep(2)


def?demo_task(url):
????"""Main?coroutine

????You?should?wrap?your?each?task?into?one?entry?coroutine,
????then?spawn?its?own?sub?coroutine?tasks.
????"""
????start_ts?=?time.time()
????r?=?urllib2.urlopen(url)
????demo_child_task()
????print('cost?{}s?for?url?{}'.format(time.time()?-?start_ts,?url))


def?main():
????start_ts?=?time.time()
????pool.map(demo_task,?TARGET_URLS)
????print('total?cost?{}s'.format(time.time()?-?start_ts))


if?__name__?==?'__main__':
????main()

tornado?demo:

#!/usr/bin/env?python
#?-*-?coding:?utf-8?-*-

"""
cost?0.5578329563140869s,?get?http://httpbin.org/get
cost?0.5621621608734131s,?get?http://httpbin.org/ip
cost?0.5613000392913818s,?get?http://httpbin.org/user-agent
cost?0.5709919929504395s,?get?http://httpbin.org/
cost?0.572376012802124s,?get?http://httpbin.org/headers
total?cost?0.5809519290924072s
"""
import?time

import?tornado
import?tornado.web
import?tornado.httpclient


TARGET_URLS?=?[
????'http://httpbin.org/',
????'http://httpbin.org/ip',
????'http://httpbin.org/user-agent',
????'http://httpbin.org/headers',
????'http://httpbin.org/get',
]


@tornado.gen.coroutine
def?demo_hanlder(ioloop):
????for?i,?url?in?enumerate(TARGET_URLS):
????????demo_task(url,?ioloop=ioloop)


@tornado.gen.coroutine
def?demo_task(url,?ioloop=None):
????start_ts?=?time.time()
????http_client?=?tornado.httpclient.AsyncHTTPClient()
????r?=?yield?http_client.fetch(url)
????#?r?is?the?response?object
????end_ts?=?time.time()
????print('cost?{}s,?get?{}'.format(end_ts?-?start_ts,?url))
????TARGET_URLS.remove(url)
????if?not?TARGET_URLS:
????????ioloop.stop()


def?main():
????start_ts?=?time.time()
????ioloop?=?tornado.ioloop.IOLoop.instance()
????ioloop.add_future(demo_hanlder(ioloop),?lambda?f:?None)
????ioloop.start()

????#?total?cost?will?equal?to?the?longest?task
????print('total?cost?{}s'.format(time.time()?-?start_ts))


if?__name__?==?'__main__':
????main()

tornado demo:

from?time?import?sleep

from?kipp.aio?import?coroutine2,?run_until_complete,?sleep,?return_in_coroutine
from?kipp.utils?import?ThreadPoolExecutor,?get_logger


executor?=?ThreadPoolExecutor(10)
logger?=?get_logger()


@coroutine2
def?coroutine_demo():
????logger.info('start?coroutine_demo')
????yield?sleep(1)
????logger.info('coroutine_demo??done')
????yield?executor.submit(blocking_func)
????return_in_coroutine('yeo')


def?blocking_func():
????logger.info('start?blocking?task...')
????sleep(1)
????logger.info('blocking?task?return')
????return?'hard'


@coroutine2
def?coroutine_main():
????logger.info('start?coroutine_main')
????r?=?yield?coroutine_demo()
????logger.info('coroutine_demo?return:?{}'.format(r))

????yield?sleep(1)
????return_in_coroutine('coroutine_main?yo')


def?main():
????f?=?coroutine_main()
????run_until_complete(f)
????logger.info('coroutine_main?return:?{}'.format(f.result()))


if?__name__?==?'__main__':
????main()

kipp demo:

from?time?import?sleep

from?kipp.aio?import?coroutine2,?run_until_complete,?sleep,?return_in_coroutine
from?kipp.utils?import?ThreadPoolExecutor,?get_logger


executor?=?ThreadPoolExecutor(10)
logger?=?get_logger()


@coroutine2
def?coroutine_demo():
????logger.info('start?coroutine_demo')
????yield?sleep(1)
????logger.info('coroutine_demo??done')
????yield?executor.submit(blocking_func)
????return_in_coroutine('yeo')


def?blocking_func():
????logger.info('start?blocking?task...')
????sleep(1)
????logger.info('blocking?task?return')
????return?'hard'


@coroutine2
def?coroutine_main():
????logger.info('start?coroutine_main')
????r?=?yield?coroutine_demo()
????logger.info('coroutine_demo?return:?{}'.format(r))

????yield?sleep(1)
????return_in_coroutine('coroutine_main?yo')


def?main():
????f?=?coroutine_main()
????run_until_complete(f)
????logger.info('coroutine_main?return:?{}'.format(f.result()))


if?__name__?==?'__main__':
????main()

使用 tornado 時(shí)需要注意，因?yàn)樗蕾?lài) generator 來(lái)模擬協(xié)程，所以函數(shù)無(wú)法返回，只能用 raise gen.Return 來(lái)模擬。3.4 里引入了 yield from 到 3.6 的 async/await 才算徹底解決了這個(gè)問(wèn)題。還有就是小心 tornado 里的 Future 不是線程安全的。

至于 gevent，容我吐個(gè)槽，求別再提 monkey_patch 了…

https://docs.python.org/3/library/asyncio-task.html 官方文檔對(duì)于 asyncio 的描述很清晰易懂，推薦一讀。一個(gè)小提示，async 函數(shù)被調(diào)用后會(huì)創(chuàng)建一個(gè) coroutine，這時(shí)候該協(xié)程并不會(huì)運(yùn)行，需要通過(guò) ensure_future 或 create_task 方法生成 Task 后才會(huì)被調(diào)度執(zhí)行。

另外，一個(gè)進(jìn)程內(nèi)不要?jiǎng)?chuàng)建多個(gè) ioloop。

做一個(gè)小結(jié)，一個(gè)簡(jiǎn)單的做法是，啟動(dòng)程序后，分別創(chuàng)建一個(gè)進(jìn)程池（進(jìn)程數(shù)小于等于可用核數(shù)）、線程池和 ioloop，ioloop 負(fù)責(zé)調(diào)度一切的協(xié)程，遇到阻塞的調(diào)用時(shí)，I/O 型的扔進(jìn)線程池，CPU 型的扔進(jìn)進(jìn)程池，這樣代碼邏輯簡(jiǎn)單，還能盡可能的利用機(jī)器性能。一個(gè)簡(jiǎn)單的完整示例：

"""
??python?process_thread_coroutine.py

[2019-08-11?09:09:37,670Z?-?INFO?-?kipp]?-?main?running...
[2019-08-11?09:09:37,671Z?-?INFO?-?kipp]?-?coroutine_main?running...
[2019-08-11?09:09:37,671Z?-?INFO?-?kipp]?-?io_blocking_task?running...
[2019-08-11?09:09:37,690Z?-?INFO?-?kipp]?-?coroutine_task?running...
[2019-08-11?09:09:37,691Z?-?INFO?-?kipp]?-?coroutine_error?running...
[2019-08-11?09:09:37,691Z?-?INFO?-?kipp]?-?coroutine_error?end,?cost?0.00s
[2019-08-11?09:09:37,693Z?-?INFO?-?kipp]?-?cpu_blocking_task?running...
[2019-08-11?09:09:38,674Z?-?INFO?-?kipp]?-?io_blocking_task?end,?cost?1.00s
[2019-08-11?09:09:38,695Z?-?INFO?-?kipp]?-?coroutine_task?end,?cost?1.00s
[2019-08-11?09:09:39,580Z?-?INFO?-?kipp]?-?cpu_blocking_task?end,?cost?1.89s
[2019-08-11?09:09:39,582Z?-?INFO?-?kipp]?-?coroutine_main?got?[None,?AttributeError('yo'),?None,?None]
[2019-08-11?09:09:39,582Z?-?INFO?-?kipp]?-?coroutine_main?end,?cost?1.91s
[2019-08-11?09:09:39,582Z?-?INFO?-?kipp]?-?main?end,?cost?1.91s
"""


from?time?import?sleep,?time
from?asyncio?import?get_event_loop,?sleep?as?asleep,?gather,?ensure_future,?iscoroutine
from?concurrent.futures?import?ProcessPoolExecutor,?ThreadPoolExecutor,?wait
from?functools?import?wraps

from?kipp.utils?import?get_logger


logger?=?get_logger()


N_FORK?=?4
N_THREADS?=?10

thread_executor?=?ThreadPoolExecutor(max_workers=N_THREADS)
process_executor?=?ProcessPoolExecutor(max_workers=N_FORK)
ioloop?=?get_event_loop()


def?timer(func):
????@wraps(func)
????def?wrapper(*args,?**kw):
????????logger.info(f"{func.__name__}?running...")
????????start_at?=?time()
????????try:
????????????r?=?func(*args,?**kw)
????????finally:
????????????logger.info(f"{func.__name__}?end,?cost?{time()?-?start_at:.2f}s")

????return?wrapper


def?async_timer(func):
????@wraps(func)
????async?def?wrapper(*args,?**kw):
????????logger.info(f"{func.__name__}?running...")
????????start_at?=?time()
????????try:
????????????return?await?func(*args,?**kw)
????????finally:
????????????logger.info(f"{func.__name__}?end,?cost?{time()?-?start_at:.2f}s")

????return?wrapper


@timer
def?io_blocking_task():
????"""I/O?型阻塞調(diào)用"""
????sleep(1)


@timer
def?cpu_blocking_task():
????"""CPU?型阻塞調(diào)用"""
????for?_?in?range(1?<26):
????????pass


@async_timer
async?def?coroutine_task():
????"""異步協(xié)程調(diào)用"""
????await?asleep(1)


@async_timer
async?def?coroutine_error():
????"""會(huì)拋出異常的協(xié)程調(diào)用"""
????raise?AttributeError("yo")


@async_timer
async?def?coroutine_main():
????ioloop?=?get_event_loop()
????r?=?await?gather(
????????coroutine_task(),
????????coroutine_error(),
????????ioloop.run_in_executor(thread_executor,?io_blocking_task),
????????ioloop.run_in_executor(process_executor,?cpu_blocking_task),
????????return_exceptions=True,
????)
????logger.info(f"coroutine_main?got?{r}")


@timer
def?main():
????get_event_loop().run_until_complete(coroutine_main())


if?__name__?==?"__main__":
????main()

學(xué)到這一步，你已經(jīng)能夠熟練的運(yùn)用協(xié)程、線程、進(jìn)程處理不同類(lèi)型的任務(wù)。接著拿上面提到的垃圾 4 核虛機(jī)舉例，你現(xiàn)在應(yīng)該可以比較輕松的實(shí)現(xiàn)達(dá)到 1k QPS 的服務(wù)，在白天十小時(shí)里可以處理超過(guò)一億請(qǐng)求，費(fèi)用依然僅 20元/天。你還有什么借口說(shuō)是因?yàn)?Python 慢呢？

人們?cè)诹牡秸Z(yǔ)言／框架／工具性能時(shí)，考慮的是“當(dāng)程序員盡可能的優(yōu)化后，工具性能會(huì)成為最終的瓶頸，所以我們一定要選一個(gè)最快的”。

但事實(shí)上是，程序員本身才是性能的最大瓶頸，而工具真正體現(xiàn)出來(lái)的價(jià)值，是在程序員很爛時(shí)，所能提供的兜底性能。

如果你覺(jué)得自己并不是那個(gè)瓶頸，那也沒(méi)必要來(lái)聽(tīng)我講了

在性能優(yōu)化上有兩句老話(huà)：

• 一定要針對(duì)瓶頸做優(yōu)化
• 過(guò)早優(yōu)化是萬(wàn)惡之源

所以我覺(jué)得要開(kāi)放、冷靜地看待工具的性能。在一套完整的業(yè)務(wù)系統(tǒng)中，框架工具往往是耗時(shí)占比最低的那個(gè)，在擴(kuò)容、緩存技術(shù)如此發(fā)達(dá)的今天，你已經(jīng)很難說(shuō)出工具性能不夠這樣的話(huà)了。

成長(zhǎng)的空間很大，多在自己身上找原因。

一個(gè)經(jīng)驗(yàn)觀察，即使在工作中不斷的實(shí)際練習(xí)，對(duì)于異步協(xié)程這種全新的思維模式，從學(xué)會(huì)到能在工作中熟練運(yùn)用且不犯大錯(cuò)，比較聰明的人也需要一個(gè)月。

換成 go 也不會(huì)好很多，await 也能實(shí)現(xiàn)同步寫(xiě)法，而且你依然需要面對(duì)我前文提到過(guò)的同步控制和資源用量?jī)蓚€(gè)核心問(wèn)題。

簡(jiǎn)單提一下性能分析，py 可以利用 cProfile、line_profiler、memory_profiler、vprof、objgraph 等工具生成耗時(shí)、內(nèi)存占用、調(diào)用關(guān)系圖、火焰圖等。

關(guān)于性能分析領(lǐng)域的更多方法論和理念，推薦閱讀《性能之巔》（過(guò)去做的關(guān)于性能之巔的部分摘抄 https://twitter.com/ppcelery/status/1051832271001382912）。

必須強(qiáng)調(diào)：優(yōu)化必須要有足夠的數(shù)據(jù)支撐，包括優(yōu)化前和優(yōu)化后。

性能優(yōu)化其實(shí)是一個(gè)非常復(fù)雜的領(lǐng)域，雖然上面提到的工具可以生成各式各樣的看上去就很厲害的圖，但是優(yōu)化不是簡(jiǎn)單的你看哪慢就去改哪，而是需要有極其扎實(shí)的基礎(chǔ)知識(shí)和全局思維的。

而且，上述工具得出的指標(biāo)，在性能尚未逼近極限時(shí)，可能會(huì)有相當(dāng)大的誤導(dǎo)性，使用的時(shí)候也要小心。

有一些較為普適的經(jīng)驗(yàn)：

• I/O 越少越好，盡量在內(nèi)存里完成
• 內(nèi)存分配越少越好，盡量復(fù)用
• 變量盡可能少，gc 友好
• 盡量提高局部性
• 盡量用內(nèi)建函數(shù)，不要輕率造輪子

下列方法如非瓶頸不要輕易用：

• 循環(huán)展開(kāi)
• 內(nèi)存對(duì)齊
• zero copy（mmap、sendfile）

測(cè)試是開(kāi)發(fā)人員很容易忽視的一個(gè)環(huán)節(jié)，很多人認(rèn)為交給 QA 即可，但其實(shí)測(cè)試也是開(kāi)發(fā)過(guò)程中的一個(gè)重要組成部分，不但可以提高軟件的交付質(zhì)量，還可以增進(jìn)你的代碼組織能力。

最常見(jiàn)的劃分可以稱(chēng)之為黑盒 & 白盒，前者是只針對(duì)接口行為的測(cè)試，后者是深入了解實(shí)現(xiàn)細(xì)節(jié)，針對(duì)實(shí)現(xiàn)方式進(jìn)行的針對(duì)性測(cè)試。

對(duì) Py 開(kāi)發(fā)者而言，最簡(jiǎn)單實(shí)用的工具就是 unitest.TestCase 和 pytest，在包內(nèi)任何以 test*.py 命名的文件，內(nèi)含 TestCase 類(lèi)的以 test* 命名的方法都會(huì)被執(zhí)行。

測(cè)試方法也很簡(jiǎn)單，你給定入?yún)?，然后調(diào)用想要測(cè)試的函數(shù)，然后檢查其返回是否符合需求，不符合就拋出異常。

https://docs.pytest.org/en/latest/

"""
test_demo.py
"""

from?unittest?import?TestCase
from?typing?import?List

def?demo(l:?List[int])?->?int:
????return?l[0]

class?DemoTestCase(TestCase):

????def?setUp(self):
????????print("first?run")

????def?tearDown(self):
????????print("last?run")

????def?test_demo(self):
????????data?=?[]
????????self.assertRaises(IndexError,?demo,?data)

開(kāi)始寫(xiě)測(cè)試后，你才會(huì)意識(shí)到你的很多函數(shù)非常難以測(cè)試。因?yàn)樗鼈兛赡苡星短渍{(diào)用，可能有內(nèi)含狀態(tài)，可能有外部依賴(lài)等等。

但是需要強(qiáng)調(diào)的是，這不但不是不寫(xiě)測(cè)試的理由，這其實(shí)正是寫(xiě)測(cè)試的目的！

通過(guò)努力地寫(xiě)測(cè)試，會(huì)強(qiáng)迫你開(kāi)始編寫(xiě)精簡(jiǎn)、功能單一、無(wú)狀態(tài)、依賴(lài)注入、避免鏈?zhǔn)秸{(diào)用的函數(shù)。

一個(gè)簡(jiǎn)單直觀的“好壞對(duì)比”，鏈?zhǔn)秸{(diào)用的函數(shù)很難測(cè)試，它內(nèi)含了太多其他函數(shù)的調(diào)用，一旦測(cè)試就變成了一個(gè)“集成測(cè)試”。而將其按照步驟一一拆分后，就可以對(duì)其進(jìn)行精細(xì)化的“單元測(cè)試”，這可以契合你開(kāi)發(fā)的步伐，步步為營(yíng)穩(wěn)步推進(jìn)。

"""
這是很糟糕的鏈?zhǔn)秸{(diào)用
"""

def?main():
????func1()


def?func1():
????return?func2()

def?func2():
????return?func3()

def?func3():
????return?"shit"



"""
這樣寫(xiě)會(huì)好很多
"""
def?step1():
????return?"yoo"


def?step2(v):
????return?f"hello,?{v}"


def?step3(v):
????return?f"you?know?nothing,?{v}"


def?main():
????r1?=?step1()
????r2?=?step2(r1)
????step3(r2)

順帶一提，對(duì)于一些無(wú)法繞開(kāi)的外部調(diào)用，如網(wǎng)絡(luò)請(qǐng)求、數(shù)據(jù)庫(kù)請(qǐng)求。單元測(cè)試的準(zhǔn)則之一就是“排除一切外部因素”，你不應(yīng)該發(fā)起任何真正的外部調(diào)用的，因?yàn)檫@會(huì)引入不可控的數(shù)據(jù)。正確做法是通過(guò)依賴(lài)注入 Mock 對(duì)象，或者通過(guò) patch 去改寫(xiě)調(diào)用的接口對(duì)象。

以前寫(xiě)過(guò)一篇簡(jiǎn)介：https://blog.laisky.com/p/unittest-mock/

單元測(cè)試應(yīng)該兼顧黑盒、白盒。你既應(yīng)該編寫(xiě)面對(duì)接口的案例，也應(yīng)該盡可能的試探內(nèi)部的實(shí)現(xiàn)路徑（增加覆蓋率）。

你還可以逐漸地把線上遇到的各種 bug 都編寫(xiě)為案例，這些案例會(huì)成為項(xiàng)目寶貴的財(cái)富，為回歸測(cè)試提供強(qiáng)有力的支持。而且有這么多測(cè)試案例提供保護(hù)，coding 的時(shí)候也會(huì)安心很多。

在單元測(cè)試的基礎(chǔ)上，人們發(fā)展出了 TDD，但是在實(shí)踐的過(guò)程中，發(fā)現(xiàn)有些“狡猾的”開(kāi)發(fā)會(huì)針對(duì)案例的特例進(jìn)行編程。為此，人們決定應(yīng)該拋棄形式，回歸本源，從方法論的高度來(lái)探尋測(cè)試的道路。其中光明一方，就是 PBT，試圖通過(guò)描述問(wèn)題的實(shí)質(zhì)，來(lái)自動(dòng)生成測(cè)試案例。

一篇簡(jiǎn)介：https://blog.laisky.com/p/pbt-hypothesis/

另一個(gè)黑暗的方向就是 Fuzzing，它干脆完全忽略函數(shù)的實(shí)現(xiàn)，貫徹黑盒到底，通過(guò)遺傳算法，隨機(jī)的生成入?yún)ⅲ詼y(cè)試到宇宙盡頭的決心，對(duì)函數(shù)進(jìn)行死纏爛打，發(fā)掘出正常人根本想不到猜不著的犄角旮旯里的 bug。

py 是一門(mén)動(dòng)態(tài)解釋型語(yǔ)言，使得你幾乎可以寫(xiě)出各種想得到的寫(xiě)法，但是能夠?qū)懞蛻?yīng)該寫(xiě)是兩回事。雖然 py 支持多樣化的寫(xiě)法，但是你還是應(yīng)該有意識(shí)的限制自己的行為，按照一定的規(guī)范進(jìn)行編碼，以盡可能的在條件允許的情況下，提高代碼的穩(wěn)健型和可維護(hù)性。

一些常見(jiàn)的規(guī)范不用多講，比如：

• 不要寫(xiě) magic value，多使用常量（如枚舉、或 XXX_VAR_NAME 這種寫(xiě)法）
• 不同類(lèi)型的參數(shù)或返回不要放在 list 里
• 盡可能多用 key-value 類(lèi)型，而不是到處都在用下標(biāo)取值
• 盡可能多用不可變類(lèi)型，函數(shù)盡可能做到冪等

此外，“靜態(tài)化”是一種提高程序可讀性和可維護(hù)性的重要手段，比如在函數(shù)定義時(shí)指明 type-hints，寫(xiě)清楚參數(shù)和返回值的類(lèi)型。以及對(duì)于 OOP，也可以寫(xiě)出定義明確的的“接口-實(shí)現(xiàn)”型代碼，比如按照 abc -> BaseClass -> Class -> Instance 的形式進(jìn)行定義，就會(huì)規(guī)范很多。

from?abc?import?ABC,?abstractmethod,?abstractproperty

class?ThingsABC(ABC):
????@abstractproperty
????def?etable(self):
????????pass


class?BaseFood(ThingsABC):
????etable?=?True


class?BirdABC(ABC):
????"""
????在抽象類(lèi)中定義抽象方法和屬性，
????實(shí)例化的時(shí)候會(huì)自動(dòng)檢查這些抽象方法和方法必須已被實(shí)現(xiàn)，否則會(huì)拋出一場(chǎng)。

????具體實(shí)現(xiàn)的方法多種多樣，比如直接在類(lèi)里定義，或者多繼承等等
????"""
????@abstractmethod
????def?fly(self):
????????pass

????@abstractmethod
????def?eat(self,?food:?BaseFood):
????????pass


class?BaseBird(BirdABC):
????"""
????可以定義一些鳥(niǎo)類(lèi)都應(yīng)該有的通用屬性和方法
????"""
????pass


class?Robin(BaseBird):
????"""
????定義一些知更鳥(niǎo)特有的屬性和方法
????"""
#?????def?fly(self):
#?????????pass

#?????def?eat(self,?foold:?BaseFood):
#?????????pass


r?=?Robin()??#?會(huì)報(bào)錯(cuò)，因?yàn)闆](méi)有實(shí)現(xiàn)抽象方法

#?---------------------------------------------------------------------------
#?TypeError?????????????????????????????????Traceback?(most?recent?call?last)
#??in?
#??????45
#??????46
#?--->?47?r?=?Robin()??#?會(huì)報(bào)錯(cuò)，因?yàn)闆](méi)有實(shí)現(xiàn)抽象方法

#?TypeError:?Can't?instantiate?abstract?class?Robin?with?abstract?methods?eat,?fly