作為在日常開發(fā)生產(chǎn)中非常實用的語言，有必要掌握一些python用法，比如爬蟲、網(wǎng)絡(luò)請求等場景，很是實用。但python是單線程的，如何提高python的處理速度，是一個很重要的問題，這個問題的一個關(guān)鍵技術(shù)，叫協(xié)程。本篇文章，講講python協(xié)程的理解與使用，主要是針對網(wǎng)絡(luò)請求這個模塊做一個梳理，希望能幫到有需要的同學(xué)。

概念篇

在理解協(xié)程這個概念及其作用場景前，先要了解幾個基本的關(guān)于操作系統(tǒng)的概念，主要是進程、線程、同步、異步、阻塞、非阻塞，了解這幾個概念，不僅是對協(xié)程這個場景，諸如消息隊列、緩存等，都有一定的幫助。接下來，編者就自己的理解和網(wǎng)上查詢的材料，做一個總結(jié)。

進程

在面試的時候，我們都會記住一個概念，進程是系統(tǒng)資源分配的最小單位。是的，系統(tǒng)由一個個程序，也就是進程組成的，一般情況下，分為文本區(qū)域、數(shù)據(jù)區(qū)域和堆棧區(qū)域。

文本區(qū)域存儲處理器執(zhí)行的代碼（機器碼），通常來說，這是一個只讀區(qū)域，防止運行的程序被意外修改。

數(shù)據(jù)區(qū)域存儲所有的變量和動態(tài)分配的內(nèi)存，又細(xì)分為初始化的數(shù)據(jù)區(qū)（所有初始化的全局、靜態(tài)、常量，以及外部變量）和為初始化的數(shù)據(jù)區(qū)（初始化為0的全局變量和靜態(tài)變量），初始化的變量最初保存在文本區(qū)，程序啟動后被拷貝到初始化的數(shù)據(jù)區(qū)。

堆棧區(qū)域存儲著活動過程調(diào)用的指令和本地變量，在地址空間里，棧區(qū)緊連著堆區(qū)，他們的增長方向相反，內(nèi)存是線性的，所以我們代碼放在低地址的地方，由低向高增長，棧區(qū)大小不可預(yù)測，隨開隨用，因此放在高地址的地方，由高向低增長。當(dāng)堆和棧指針重合的時候，意味著內(nèi)存耗盡，造成內(nèi)存溢出。

進程的創(chuàng)建和銷毀都是相對于系統(tǒng)資源，非常消耗資源，是一種比較昂貴的操作。進程為了自身能得到運行，必須要搶占式的爭奪CPU。對于單核CPU來說，在同一時間只能執(zhí)行一個進程的代碼，所以在單核CPU上實現(xiàn)多進程，是通過CPU快速的切換不同進程，看上去就像是多個進程在同時進行。

由于進程間是隔離的，各自擁有自己的內(nèi)存內(nèi)存資源，相比于線程的共同共享內(nèi)存來說，相對安全，不同進程之間的數(shù)據(jù)只能通過 IPC（Inter-Process Communication） 進行通信共享。

線程

線程是CPU調(diào)度的最小單位。如果進程是一個容器，線程就是運行在容器里面的程序，線程是屬于進程的，同個進程的多個線程共享進程的內(nèi)存地址空間。

線程間的通信可以直接通過全局變量進行通信，所以相對來說，線程間通信是不太安全的，因此引入了各種鎖的場景，不在這里闡述。

當(dāng)一個線程崩潰了，會導(dǎo)致整個進程也崩潰了，即其他線程也掛了， 但多進程而不會，一個進程掛了，另一個進程依然照樣運行。

在多核操作系統(tǒng)中，默認(rèn)進程內(nèi)只有一個線程，所以對多進程的處理就像是一個進程一個核心。

同步和異步

同步和異步關(guān)注的是消息通信機制，所謂同步，就是在發(fā)出一個函數(shù)調(diào)用時，在沒有得到結(jié)果之前，該調(diào)用不會返回。一旦調(diào)用返回，就立即得到執(zhí)行的返回值，即調(diào)用者主動等待調(diào)用結(jié)果。所謂異步，就是在請求發(fā)出去后，這個調(diào)用就立即返回，沒有返回結(jié)果，通過回調(diào)等方式告知該調(diào)用的實際結(jié)果。

同步的請求，需要主動讀寫數(shù)據(jù)，并且等待結(jié)果;異步的請求，調(diào)用者不會立刻得到結(jié)果。而是在調(diào)用發(fā)出后，被調(diào)用者通過狀態(tài)、通知來通知調(diào)用者，或通過回調(diào)函數(shù)處理這個調(diào)用。

阻塞和非阻塞

關(guān)注的是程序在等待調(diào)用結(jié)果（消息，返回值）時的狀態(tài)。

阻塞調(diào)用是指調(diào)用結(jié)果返回之前，當(dāng)前線程會被掛起。調(diào)用線程只有在得到結(jié)果之后才會返回。非阻塞調(diào)用指在不能立刻得到結(jié)果之前，該調(diào)用不會阻塞當(dāng)前線程。所以，區(qū)分的條件在于，進程/線程要訪問的數(shù)據(jù)是否就緒，進程/線程是否需要等待。

非阻塞一般通過多路復(fù)用實現(xiàn)，多路復(fù)用有 select、poll、epoll幾種實現(xiàn)方式。

協(xié)程

在了解前面的幾個概念后，我們再來看協(xié)程的概念。

協(xié)程是屬于線程的，又稱微線程，纖程，英文名Coroutine。舉個例子，在執(zhí)行函數(shù)A時，我希望隨時中斷去執(zhí)行函數(shù)B，然后中斷B的執(zhí)行，切換回來執(zhí)行A。這就是協(xié)程的作用，由調(diào)用者自由切換。這個切換過程并不是等同于函數(shù)調(diào)用，因為它沒有調(diào)用語句。執(zhí)行方式與多線程類似，但是協(xié)程只有一個線程執(zhí)行。

協(xié)程的優(yōu)點是執(zhí)行效率非常高，因為協(xié)程的切換由程序自身控制，不需要切換線程，即沒有切換線程的開銷。同時，由于只有一個線程，不存在沖突問題，不需要依賴鎖（加鎖與釋放鎖存在很多資源消耗）。

協(xié)程主要的使用場景在于處理IO密集型程序，解決效率問題，不適用于CPU密集型程序的處理。然而實際場景中這兩種場景非常多，如果要充分發(fā)揮CPU利用率，可以結(jié)合多進程+協(xié)程的方式。后續(xù)我們會講到結(jié)合點。

原理篇

根據(jù)wikipedia的定義，協(xié)程是一個無優(yōu)先級的子程序調(diào)度組件，允許子程序在特點的地方掛起恢復(fù)。所以理論上，只要內(nèi)存足夠，一個線程中可以有任意多個協(xié)程，但同一時刻只能有一個協(xié)程在運行，多個協(xié)程分享該線程分配到的計算機資源。協(xié)程是為了充分發(fā)揮異步調(diào)用的優(yōu)勢，異步操作則是為了避免IO操作阻塞線程。

知識準(zhǔn)備

在了解原理前，我們先做一個知識的準(zhǔn)備工作。

1）現(xiàn)代主流的操作系統(tǒng)幾乎都是分時操作系統(tǒng)，即一臺計算機采用時間片輪轉(zhuǎn)的方式為多個用戶服務(wù)，系統(tǒng)資源分配的基本單位是進程，CPU調(diào)度的基本單位是線程。

2）運行時內(nèi)存空間分為變量區(qū)，棧區(qū)，堆區(qū)。內(nèi)存地址分配上，堆區(qū)從低地到高，棧區(qū)從高往低。

3）計算機執(zhí)行時一條條指令讀取執(zhí)行，執(zhí)行到當(dāng)前指令時，下一條指令的地址在指令寄存器的IP中，ESP寄存值指向當(dāng)前棧頂?shù)刂罚珽BP指向當(dāng)前活動棧幀的基地址。

4）系統(tǒng)發(fā)生函數(shù)調(diào)用時操作為：先將入?yún)挠彝笠来螇簵＃缓蟀逊祷氐刂穳簵＃詈髮?dāng)前EBP寄存器的值壓棧，修改ESP寄存器的值，在棧區(qū)分配當(dāng)前函數(shù)局部變量所需的空間。

5）協(xié)程的上下文包含屬于當(dāng)前協(xié)程的棧區(qū)和寄存器里面存放的值。

事件循環(huán)

在python3.3中，通過關(guān)鍵字yield from使用協(xié)程，在3.5中，引入了關(guān)于協(xié)程的語法糖async和await，我們主要看async/await的原理解析。其中，事件循環(huán)是一個核心所在，編寫過 js的同學(xué)，會對事件循環(huán)Eventloop更加了解， 事件循環(huán)是一種等待程序分配事件或消息的編程架構(gòu)（維基百科）。在python中，asyncio.coroutine 修飾器用來標(biāo)記作為協(xié)程的函數(shù)， 這里的協(xié)程是和asyncio及其事件循環(huán)一起使用的，而在后續(xù)的發(fā)展中，async/await被使用的越來越廣泛。

async/await

async/await是使用python協(xié)程的關(guān)鍵，從結(jié)構(gòu)上來看，asyncio 實質(zhì)上是一個異步框架，async/await 是為異步框架提供的 API已方便使用者調(diào)用，所以使用者要想使用async/await 編寫協(xié)程代碼，目前必須機遇 asyncio 或其他異步庫。

Future

在實際開發(fā)編寫異步代碼時，為了避免太多的回調(diào)方法導(dǎo)致的回調(diào)地獄，但又需要獲取異步調(diào)用的返回結(jié)果結(jié)果，聰明的語言設(shè)計者設(shè)計了一個 叫Future的對象，封裝了與loop 的交互行為。其大致執(zhí)行過程為：程序啟動后，通過add_done_callback 方法向 epoll 注冊回調(diào)函數(shù)，當(dāng) result 屬性得到返回值后，主動運行之前注冊的回調(diào)函數(shù)，向上傳遞給 coroutine。這個Future對象為asyncio.Future。

但是，要想取得返回值，程序必須恢復(fù)恢復(fù)工作狀態(tài)，而由于Future 對象本身的生存周期比較短，每一次注冊回調(diào)、產(chǎn)生事件、觸發(fā)回調(diào)過程后工作可能已經(jīng)完成，所以用 Future 向生成器 send result 并不合適。所以這里又引入一個新的對象 Task，保存在Future 對象中，對生成器協(xié)程進行狀態(tài)管理。

Python 里另一個 Future 對象是 concurrent.futures.Future，與 asyncio.Future 互不兼容，容易產(chǎn)生混淆。區(qū)別點在于，concurrent.futures 是線程級的 Future 對象，當(dāng)使用 concurrent.futures.Executor 進行多線程編程時，該對象用于在不同的 thread 之間傳遞結(jié)果。

Task

上文中提到，Task是維護生成器協(xié)程狀態(tài)處理執(zhí)行邏輯的的任務(wù)對象，Task 中有一個_step 方法，負(fù)責(zé)生成器協(xié)程與 EventLoop 交互過程的狀態(tài)遷移，整個過程可以理解為：Task向協(xié)程 send 一個值，恢復(fù)其工作狀態(tài)。當(dāng)協(xié)程運行到斷點后，得到新的Future對象，再處理 future 與 loop 的回調(diào)注冊過程。

Loop

在日常開發(fā)中，會有一個誤區(qū)，認(rèn)為每個線程都可以有一個獨立的 loop。實際運行時，主線程才能通過 asyncio.get_event_loop（） 創(chuàng)建一個新的 loop，而在其他線程時，使用 get_event_loop（） 卻會拋錯。正確的做法為通過 asyncio.set_event_loop（） ，將當(dāng)前線程與 主線程的loop 顯式綁定。

Loop有一個很大的缺陷，就是 loop 的運行狀態(tài)不受 Python 代碼控制，所以在業(yè)務(wù)處理中，無法穩(wěn)定的將協(xié)程拓展到多線程中運行。

總結(jié)

實戰(zhàn)篇

介紹完概念和原理，我來看看如何使用，這里，舉一個實際場景的例子，來看看如何使用python的協(xié)程。

場景

外部接收一些文件，每個文件里有一組數(shù)據(jù)，其中，這組數(shù)據(jù)需要通過http的方式，發(fā)向第三方平臺，并獲得結(jié)果。

分析

由于同一個文件的每一組數(shù)據(jù)沒有前后的處理邏輯，在之前通過Requests庫發(fā)送的網(wǎng)絡(luò)請求，串行執(zhí)行，下一組數(shù)據(jù)的發(fā)送需要等待上一組數(shù)據(jù)的返回，顯得整個文件的處理時間長，這種請求方式，完全可以由協(xié)程來實現(xiàn)。

為了更方便的配合協(xié)程發(fā)請求，我們使用aiohttp庫來代替requests庫，關(guān)于aiohttp，這里不做過多剖析，僅做下簡單介紹。

aiohttp

aiohttp是asyncio和Python的異步HTTP客戶端/服務(wù)器，由于是異步的，經(jīng)常用在服務(wù)區(qū)端接收請求，和客戶端爬蟲應(yīng)用，發(fā)起異步請求，這里我們主要用來發(fā)請求。

aiohttp支持客戶端和HTTP服務(wù)器，可以實現(xiàn)單線程并發(fā)IO操作，無需使用Callback Hell即可支持Server WebSockets和Client WebSockets，且具有中間件。

代碼實現(xiàn)

直接上代碼了，talk is cheap， show me the code~

import aiohttp

import asyncio

from inspect import isfunction

import time

import logger

@logging_utils.exception（logger）

def request（pool， data_list）：

loop = asyncio.get_event_loop（）

loop.run_until_complete（exec（pool， data_list））

async def exec（pool， data_list）：

tasks = ［］

sem = asyncio.Semaphore（pool）

for item in data_list：

tasks.append（

control_sem（sem，

item.get（“method”， “GET”），

item.get（“url”），

item.get（“data”），

item.get（“headers”），

item.get（“callback”）））

await asyncio.wait（tasks）

async def control_sem（sem， method， url， data， headers， callback）：

async with sem：

count = 0

flag = False

while not flag and count 《 4：

flag = await fetch（method， url， data， headers， callback）

count = count + 1

print（“flag：{}，count：{}”.format（flag， count））

if count == 4 and not flag：

raise Exception（‘EAS service not responding after 4 times of retry.’）

async def fetch（method， url， data， headers， callback）：

async with aiohttp.request（method， url=url， data=data， headers=headers） as resp：

try：

json = await resp.read（）

print（json）

if resp.status ！= 200：

return False

if isfunction（callback）：

callback（json）

return True

except Exception as e：

print（e）

這里，我們封裝了對外發(fā)送批量請求的request方法，接收一次性發(fā)送的數(shù)據(jù)多少，和數(shù)據(jù)綜合，在外部使用時，只需要構(gòu)建好網(wǎng)絡(luò)請求對象的數(shù)據(jù)，設(shè)定好請求池大小即可，同時，設(shè)置了重試功能，進行了4次重試，防止在網(wǎng)絡(luò)抖動的時候，單個數(shù)據(jù)的網(wǎng)絡(luò)請求發(fā)送失敗。

最終效果

在使用協(xié)程重構(gòu)網(wǎng)絡(luò)請求模塊之后，當(dāng)數(shù)據(jù)量在1000的時候，由之前的816s，提升到424s，快了一倍，且請求池大小加大的時候，效果更明顯，由于第三方平臺同時建立連接的數(shù)據(jù)限制，我們設(shè)定了40的閥值。可以看到，優(yōu)化的程度很顯著。

責(zé)任編輯：ct