[[358307]]

 1. 協(xié)程相關(guān)的概念

1.1 進(jìn)程和線程

進(jìn)程（Process）是應(yīng)用程序啟動的實(shí)例，擁有代碼、數(shù)據(jù)和文件和獨(dú)立的內(nèi)存空間，是操作系統(tǒng)最小資源管理單元。每個進(jìn)程下面有一個或者多個線程（Thread），來負(fù)責(zé)執(zhí)行程序的計算，是最小的執(zhí)行單元。

重點(diǎn)是：操作系統(tǒng)會負(fù)責(zé)進(jìn)程的資源的分配；控制權(quán)主要在操作系統(tǒng)。另一方面，線程做為任務(wù)的執(zhí)行單元，有新建、可運(yùn)行runnable（調(diào)用start方法，進(jìn)入調(diào)度池，等待獲取cpu使用權(quán)）、運(yùn)行running（得到cpu使用權(quán)開始執(zhí)行程序） 阻塞blocked（放棄了cpu 使用權(quán)，再次等待） 死亡dead5中不同的狀態(tài)。線程的轉(zhuǎn)態(tài)也是由操作系統(tǒng)進(jìn)行控制。線程如果存在資源共享的情況下，就需要加鎖，比如生產(chǎn)者和消費(fèi)者模式，生產(chǎn)者生產(chǎn)數(shù)據(jù)多共享隊(duì)列，消費(fèi)者從共享隊(duì)列中消費(fèi)數(shù)據(jù)。

線程和進(jìn)程在得到和放棄cpu使用權(quán)時，cpu使用權(quán)的切換都需損耗性能，因?yàn)槟硞€線程為了能夠在再次獲得cpu使用權(quán)時能繼續(xù)執(zhí)行任務(wù)，必須記住上一次執(zhí)行的所有狀態(tài)。另外線程還有鎖的問題。

1.2 并行和并發(fā)

并行和并發(fā)，聽起來都像是同時執(zhí)行不同的任務(wù)。但是這個同時的含義是不一樣的。

•  并行：多核CPU才有可能真正的同時執(zhí)行，就是獨(dú)立的資源來完成不同的任務(wù)，沒有先后順序。
•  并發(fā)（concurrent）：是看上去的同時執(zhí)行，實(shí)際微觀層面是順序執(zhí)行，是操作系統(tǒng)對進(jìn)程的調(diào)度以及cpu的快速上下文切換，每個進(jìn)程執(zhí)行一會然后停下來，cpu資源切換到另一個進(jìn)程，只是切換的時間很短，看起來是多個任務(wù)同時在執(zhí)行。要實(shí)現(xiàn)大并發(fā)，需要把任務(wù)切成小的任務(wù)。

上面說的多核cpu可能同時執(zhí)行，這里的可能是和操作系統(tǒng)調(diào)度有關(guān)，如果操作系統(tǒng)調(diào)度到同一個cpu，那就需要cpu進(jìn)行上下文切換。當(dāng)然多核情況下，操作系統(tǒng)調(diào)度會盡可能考慮不同cpu。

這里的上下文切換可以理解為需要保留不同執(zhí)行任務(wù)的狀態(tài)和數(shù)據(jù)。所有的并發(fā)處理都有排隊(duì)等候，喚醒，執(zhí)行至少三個這樣的步驟

1.3 協(xié)程

我們知道線程的提出是為了能夠在多核cpu的情況下，達(dá)到并行的目的。而且線程的執(zhí)行完全是操作系統(tǒng)控制的。而協(xié)程（Coroutine）是線程下的，控制權(quán)在于用戶，本質(zhì)是為了能讓多組過程能不獨(dú)自占用完所有資源，在一個線程內(nèi)交叉執(zhí)行，達(dá)到高并發(fā)的目的。

協(xié)程的優(yōu)勢：

•  協(xié)程最大的優(yōu)勢就是協(xié)程極高的執(zhí)行效率。因?yàn)樽映绦蚯袚Q不是線程切換，而是由程序自身控制，因此，沒有線程切換的開銷，和多線程比，線程數(shù)量越多，協(xié)程的性能優(yōu)勢就越明顯
•  第二大優(yōu)勢就是不需要多線程的鎖機(jī)制，因?yàn)橹挥幸粋€線程，也不存在同時寫變量沖突，在協(xié)程中控制共享資源不加鎖，只需要判斷狀態(tài)就好了，所以執(zhí)行效率比多線程高很多。

協(xié)程和線程區(qū)別：

•  協(xié)程都沒參與多核CPU并行處理，協(xié)程是不并行
•  線程在多核處理器上是并行在單核處理器是受操作系統(tǒng)調(diào)度的
•  協(xié)程需要保留上一次調(diào)用的狀態(tài)
•  線程的狀態(tài)有操作系統(tǒng)來控制

我們姑且也過一遍這些文字上的概念，show your code的時候再聯(lián)系起來，就會更清晰的。

2. python中的線程

python中的線程由于歷史原因，即使在多核cpu的情況下并不能達(dá)真正的并行。這個原因就是全局解釋器鎖GIL（global interpreter lock），準(zhǔn)確的說GIL不是python的特性，而是cpython引入的一個概念。cpython解釋器在解析多線程時，會上GIL鎖，保證同一時刻只有一個線程獲取CPU使用權(quán)。

•  為什么需要GIL python中一切都是對象，Cpython中對象的回收，是通過對象的引用計數(shù)來判斷，當(dāng)對象的引用計數(shù)為0時，就會進(jìn)行垃圾回收，自動釋放內(nèi)存。但是如果多線程的情況，引用計數(shù)就變成了一個共享的變量 Cpython是當(dāng)下最流行的Python的解釋器，使用引用計數(shù)來管理內(nèi)存，在Python中，一切都是對象，引用計數(shù)就是指向?qū)ο蟮闹羔様?shù)，當(dāng)這個數(shù)字變成0，則會進(jìn)行垃圾回收，自動釋放內(nèi)存。但是問題是Cpython是線程不安全的。

考慮下如果有兩個線程A和B同時引用一個對象obj，這個時候obj的引用計數(shù)為2；A打算撤銷對obj的引用，完成第一步時引用計數(shù)減去1時，這時發(fā)生了線程切換，A掛起等待，還沒執(zhí)行銷毀對象操作。B進(jìn)入運(yùn)行狀態(tài)，這個時候B也對obj撤銷引用，并完成引用計數(shù)減1，銷毀對象，這個時候obj的引用數(shù)為0，釋放內(nèi)存。如果此時A重新喚醒，要繼續(xù)銷毀對象，可是這個時候已經(jīng)沒有對象了。所以為了保證不出現(xiàn)數(shù)據(jù)污染，才引入GIL。

每個線程使用前都會去獲取GIL權(quán)限，使用完釋放GIL權(quán)限。釋放線程的時機(jī)由python的另一個機(jī)制check_interval來決定。

在多核cpu時，因?yàn)樾枰@取和釋放GIL鎖，會存在性能上額外的損耗。特別是由于調(diào)度控制的原因，比如一個線程釋放了鎖，調(diào)度接著又分配cpu資源給同一個線程，該線程發(fā)起申請時，又重新獲得GIL，而其他線程實(shí)際上都在等待，白白浪費(fèi)了申請和釋放鎖的操作耗時。

python中的線程比較適合I/O密集型的操作（磁盤IO或者網(wǎng)絡(luò)IO）。

•  線程的使用 

import os  
import time  
import sys  
from concurrent import futures  
def to_do(info):    
    for i in range(100000000):  
        pass  
    return info[0]  
MAX_WORKERS = 10  
param_list = []  
for i in range(5):  
    param_list.append(('text%s' % i, 'info%s' % i))  
workers = min(MAX_WORKERS, len(param_list))  
# with 默認(rèn)會等所有任務(wù)都完成才返回，所以這里會阻塞  
with futures.ThreadPoolExecutor(workers) as executor:  
    results = executor.map(to_do, sorted(param_list))  
# 打印所有  
for result in results:  
    print(result)  
# 非阻塞的方式，適合不需要返回結(jié)果的情況  
workers = min(MAX_WORKERS, len(param_list))  
executor = futures.ThreadPoolExecutor(workers)  
results = []  
for idx, param in enumerate(param_list):  
    result = executor.submit(to_do, param)  
    results.append(result)  
    print('result %s' % idx)  
# 手動等待所有任務(wù)完成  
executor.shutdown()  
print('='*10)  
for result in results:  
    print(result.result()) 

3. python中的進(jìn)程

python提供的multiprocessing包來規(guī)避GIL的缺點(diǎn)，實(shí)現(xiàn)在多核cpu上并行的目的。multiprocessing還提供進(jìn)程之間數(shù)據(jù)和內(nèi)存共享的機(jī)制。這里介紹的concurrent.futures的實(shí)現(xiàn)。用法和線程基本一樣，ThreadPoolExecutor改成ProcessPoolExecutor 

import os  
import time  
import sys  
from concurrent import futures  
def to_do(info):    
    for i in range(10000000):  
        pass  
    return info[0]  
start_time = time.time()  
MAX_WORKERS = 10  
param_list = []  
for i in range(5):  
    param_list.append(('text%s' % i, 'info%s' % i))  
workers = min(MAX_WORKERS, len(param_list))  
# with 默認(rèn)會等所有任務(wù)都完成才返回，所以這里會阻塞  
with futures.ProcessPoolExecutor(workers) as executor:  
    results = executor.map(to_do, sorted(param_list))    
 # 打印所有  
for result in results:  
    print(result)  
print(time.time()-start_time)  
# 耗時0.3704512119293213s， 而線程版本需要14.935384511947632s 

4. python中的協(xié)程

4.1 簡單協(xié)程

我們先來看下python是怎么實(shí)現(xiàn)協(xié)程的。答案是yield。以下例子的功能是實(shí)現(xiàn)計算移動平均數(shù) 

from collections import namedtuple  
Result = namedtuple('Result', 'count average')  
# 協(xié)程函數(shù)  
def averager():  
    total = 0.0  
    count = 0  
    average = None  
    while True:  
        term = yield None  # 暫停，等待主程序傳入數(shù)據(jù)喚醒  
        if term is None:  
            break  # 決定是否退出  
        total += term  
        count += 1  
        average = total/count # 累計狀態(tài)，包括上一次的狀態(tài)  
    return Result(count, average)  
# 協(xié)程的觸發(fā)  
coro_avg = averager()  
# 預(yù)激活協(xié)程  
next(coro_avg)  
# 調(diào)用者給協(xié)程提供數(shù)據(jù)  
coro_avg.send(10)  
coro_avg.send(30)  
coro_avg.send(6.5)  
try:  
    coro_avg.send(None)  
except StopIteration as exc: # 執(zhí)行完成，會拋出StopIteration異常，返回值包含在異常的屬性value里  
    result = exc.value  
print(result) 

yield關(guān)鍵字有兩個含義：產(chǎn)出和讓步;  把yield的右邊的值產(chǎn)出給調(diào)用方，同時做出讓步，暫停執(zhí)行，讓程序繼續(xù)執(zhí)行。

上面的例子可知

•  協(xié)程用yield來控制流程，接收和產(chǎn)出數(shù)據(jù)
•  next()：預(yù)激活協(xié)程
•  send：協(xié)程從調(diào)用方接收數(shù)據(jù)
•  StopIteration：控制協(xié)程結(jié)束， 同時獲取返回值

我們來回顧下1.3中協(xié)程的概念：本質(zhì)是為了能讓多組過程能不獨(dú)自占用完所有資源，在一個線程內(nèi)交叉執(zhí)行，達(dá)到高并發(fā)的目的。。上面的例子怎么解釋呢？

•  可以把一個協(xié)程單次一個任務(wù)，即移動平均
•  每個任務(wù)可以拆分成小步驟（也可以說是子程序), 即每次算一個數(shù)的平均
•  如果多個任務(wù)需要執(zhí)行呢？怎么調(diào)用控制器在調(diào)用方
•   如果有10個，可以想象，調(diào)用在控制的時候隨機(jī)的給每個任務(wù)send的一個數(shù)據(jù)化，就會是多個任務(wù)在交叉執(zhí)行，達(dá)到并發(fā)的目的。

4.2 asyncio協(xié)程應(yīng)用包

asyncio即異步I/O, 如在高并發(fā)（如百萬并發(fā)）網(wǎng)絡(luò)請求。異步I/O即你發(fā)起一個I/O操作不必等待執(zhí)行結(jié)束，可以做其他事情。asyncio底層是協(xié)程的方式來實(shí)現(xiàn)的。我們先來看一個例子，了解下asyncio的五臟六腑。 

import time  
import asyncio  
now = lambda : time.time()  
# async定義協(xié)程  
async def do_some_work(x):  
    print("waiting:",x)  
    # await掛起阻塞， 相當(dāng)于yield， 通常是耗時操作  
    await asyncio.sleep(x)  
    return "Done after {}s".format(x)  
# 回調(diào)函數(shù)，和yield產(chǎn)出類似功能  
def callback(future):  
    print("callback:",future.result())  
start = now()  
tasks = []  
for i in range(1, 4):  
    # 定義多個協(xié)程，同時預(yù)激活  
    coroutine = do_some_work(i)  
    task = asyncio.ensure_future(coroutine)  
    task.add_done_callback(callback)  
    tasks.append(task)  
# 定一個循環(huán)事件列表，把任務(wù)協(xié)程放在里面，  
loop = asyncio.get_event_loop()  
try: 
     # 異步執(zhí)行協(xié)程，直到所有操作都完成， 也可以通過asyncio.gather來收集多個任務(wù)  
    loop.run_until_complete(asyncio.wait(tasks))  
    for task in tasks:  
        print("Task ret:",task.result())  
except KeyboardInterrupt as e: # 協(xié)程任務(wù)的狀態(tài)控制  
    print(asyncio.Task.all_tasks())  
    for task in asyncio.Task.all_tasks():  
        print(task.cancel())  
    loop.stop()  
    loop.run_forever()  
finally:  
    loop.close()  
print("Time:", now()-start) 

上面涉及到的幾個概念：

•  event_loop 事件循環(huán)：程序開啟一個無限循環(huán)，把一些函數(shù)注冊到事件循環(huán)上，當(dāng)滿足事件發(fā)生的時候，調(diào)用相應(yīng)的協(xié)程函數(shù)
•  coroutine 協(xié)程：協(xié)程對象，指一個使用async關(guān)鍵字定義的函數(shù)，它的調(diào)用不會立即執(zhí)行函數(shù)，而是會返回一個協(xié)程對象。協(xié)程對象需要注冊到事件循環(huán)，由事件循環(huán)調(diào)用。
•  task任務(wù)：一個協(xié)程對象就是一個原生可以掛起的函數(shù)，任務(wù)則是對協(xié)程進(jìn)一步封裝，其中包含了任務(wù)的各種狀態(tài)
•  future: 代表將來執(zhí)行或沒有執(zhí)行的任務(wù)的結(jié)果。它和task上沒有本質(zhì)上的區(qū)別
•  async/await 關(guān)鍵字：python3.5用于定義協(xié)程的關(guān)鍵字，async定義一個協(xié)程，await用于掛起阻塞的異步調(diào)用接口。從上面可知，asyncio通過事件的方式幫我們實(shí)現(xiàn)了協(xié)程調(diào)用方的控制權(quán)處理，包括send給協(xié)程數(shù)據(jù)等。我們只要通過async定義協(xié)程，await定義阻塞，然后封裝成future的task，放入循環(huán)的事件列表中，就等著返回數(shù)據(jù)。

再來看一個http下載的例子，比如你想下載5個不同的url（同樣的，你想接收外部的百萬的請求） 

import time  
import asyncio  
from aiohttp import ClientSession  
tasks = []  
url = "https://www.baidu.com/{}"  
async def hello(url):  
    async with ClientSession() as session: 
        async with session.get(url) as response:  
            response = await response.read()  
#            print(response)  
            print('Hello World:%s' % time.time()) 
if __name__ == '__main__':  
    loop = asyncio.get_event_loop()  
    for i in range(5):  
        task = asyncio.ensure_future(hello(url.format(i)))  
        tasks.append(task)  
    loop.run_until_complete(asyncio.wait(tasks)) 

4.3 協(xié)程的應(yīng)用場景

•  支撐高并發(fā)I/O情況，如寫支撐高并發(fā)的服務(wù)端
•  代替線程，提供并發(fā)性能
•  tornado和gevent都實(shí)現(xiàn)了類似功能， 之前文章提到Twisted也是

5. 總結(jié)

本文分享關(guān)于python協(xié)程的概念和asyncio包的初步使用情況，同時也介紹了基本的相關(guān)概念，如進(jìn)程、線程、并發(fā)、并行等。希望對你有幫助，歡迎交流（@mintel）。簡要總結(jié)如下：

•  并發(fā)和并行不一樣，并行是同時執(zhí)行多個任務(wù)， 并發(fā)是在極短時間內(nèi)處理多個任務(wù)
•  多核cpu，進(jìn)程是并行，python線程受制于GIL，不能并行，反而因?yàn)樯舷挛那袚Q更耗時，協(xié)程正好可以彌補(bǔ)
•  協(xié)程也不是并行，只是任務(wù)交替執(zhí)行任務(wù)，在存在阻塞I/O情況，能夠異步執(zhí)行，提高效率
•  asyncio 異步I/O庫，可用于開發(fā)高并發(fā)應(yīng)用