[[358307]]

 1. 協程相關的概念

1.1 進程和線程

進程（Process）是應用程序啟動的實例，擁有代碼、數據和文件和獨立的內存空間，是操作系統最小資源管理單元。每個進程下面有一個或者多個線程（Thread），來負責執行程序的計算，是最小的執行單元。

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

線程和進程在得到和放棄cpu使用權時，cpu使用權的切換都需損耗性能，因為某個線程為了能夠在再次獲得cpu使用權時能繼續執行任務，必須記住上一次執行的所有狀態。另外線程還有鎖的問題。

1.2 并行和并發

并行和并發，聽起來都像是同時執行不同的任務。但是這個同時的含義是不一樣的。

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

上面說的多核cpu可能同時執行，這里的可能是和操作系統調度有關，如果操作系統調度到同一個cpu，那就需要cpu進行上下文切換。當然多核情況下，操作系統調度會盡可能考慮不同cpu。

這里的上下文切換可以理解為需要保留不同執行任務的狀態和數據。所有的并發處理都有排隊等候，喚醒，執行至少三個這樣的步驟

1.3 協程

我們知道線程的提出是為了能夠在多核cpu的情況下，達到并行的目的。而且線程的執行完全是操作系統控制的。而協程（Coroutine）是線程下的，控制權在于用戶，本質是為了能讓多組過程能不獨自占用完所有資源，在一個線程內交叉執行，達到高并發的目的。

協程的優勢：

•  協程最大的優勢就是協程極高的執行效率。因為子程序切換不是線程切換，而是由程序自身控制，因此，沒有線程切換的開銷，和多線程比，線程數量越多，協程的性能優勢就越明顯
•  第二大優勢就是不需要多線程的鎖機制，因為只有一個線程，也不存在同時寫變量沖突，在協程中控制共享資源不加鎖，只需要判斷狀態就好了，所以執行效率比多線程高很多。

協程和線程區別：

•  協程都沒參與多核CPU并行處理，協程是不并行
•  線程在多核處理器上是并行在單核處理器是受操作系統調度的
•  協程需要保留上一次調用的狀態
•  線程的狀態有操作系統來控制

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

2. python中的線程

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

•  為什么需要GIL python中一切都是對象，Cpython中對象的回收，是通過對象的引用計數來判斷，當對象的引用計數為0時，就會進行垃圾回收，自動釋放內存。但是如果多線程的情況，引用計數就變成了一個共享的變量 Cpython是當下最流行的Python的解釋器，使用引用計數來管理內存，在Python中，一切都是對象，引用計數就是指向對象的指針數，當這個數字變成0，則會進行垃圾回收，自動釋放內存。但是問題是Cpython是線程不安全的。

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

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

在多核cpu時，因為需要獲取和釋放GIL鎖，會存在性能上額外的損耗。特別是由于調度控制的原因，比如一個線程釋放了鎖，調度接著又分配cpu資源給同一個線程，該線程發起申請時，又重新獲得GIL，而其他線程實際上都在等待，白白浪費了申請和釋放鎖的操作耗時。

python中的線程比較適合I/O密集型的操作（磁盤IO或者網絡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 默認會等所有任務都完成才返回，所以這里會阻塞  
with futures.ThreadPoolExecutor(workers) as executor:  
    results = executor.map(to_do, sorted(param_list))  
# 打印所有  
for result in results:  
    print(result)  
# 非阻塞的方式，適合不需要返回結果的情況  
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)  
# 手動等待所有任務完成  
executor.shutdown()  
print('='*10)  
for result in results:  
    print(result.result()) 

3. python中的進程

python提供的multiprocessing包來規避GIL的缺點，實現在多核cpu上并行的目的。multiprocessing還提供進程之間數據和內存共享的機制。這里介紹的concurrent.futures的實現。用法和線程基本一樣，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 默認會等所有任務都完成才返回，所以這里會阻塞  
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中的協程

4.1 簡單協程

我們先來看下python是怎么實現協程的。答案是yield。以下例子的功能是實現計算移動平均數 

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

yield關鍵字有兩個含義：產出和讓步;  把yield的右邊的值產出給調用方，同時做出讓步，暫停執行，讓程序繼續執行。

上面的例子可知

•  協程用yield來控制流程，接收和產出數據
•  next()：預激活協程
•  send：協程從調用方接收數據
•  StopIteration：控制協程結束， 同時獲取返回值

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

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

4.2 asyncio協程應用包

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

import time  
import asyncio  
now = lambda : time.time()  
# async定義協程  
async def do_some_work(x):  
    print("waiting:",x)  
    # await掛起阻塞， 相當于yield， 通常是耗時操作  
    await asyncio.sleep(x)  
    return "Done after {}s".format(x)  
# 回調函數，和yield產出類似功能  
def callback(future):  
    print("callback:",future.result())  
start = now()  
tasks = []  
for i in range(1, 4):  
    # 定義多個協程，同時預激活  
    coroutine = do_some_work(i)  
    task = asyncio.ensure_future(coroutine)  
    task.add_done_callback(callback)  
    tasks.append(task)  
# 定一個循環事件列表，把任務協程放在里面，  
loop = asyncio.get_event_loop()  
try: 
     # 異步執行協程，直到所有操作都完成， 也可以通過asyncio.gather來收集多個任務  
    loop.run_until_complete(asyncio.wait(tasks))  
    for task in tasks:  
        print("Task ret:",task.result())  
except KeyboardInterrupt as e: # 協程任務的狀態控制  
    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 事件循環：程序開啟一個無限循環，把一些函數注冊到事件循環上，當滿足事件發生的時候，調用相應的協程函數
•  coroutine 協程：協程對象，指一個使用async關鍵字定義的函數，它的調用不會立即執行函數，而是會返回一個協程對象。協程對象需要注冊到事件循環，由事件循環調用。
•  task任務：一個協程對象就是一個原生可以掛起的函數，任務則是對協程進一步封裝，其中包含了任務的各種狀態
•  future: 代表將來執行或沒有執行的任務的結果。它和task上沒有本質上的區別
•  async/await 關鍵字：python3.5用于定義協程的關鍵字，async定義一個協程，await用于掛起阻塞的異步調用接口。從上面可知，asyncio通過事件的方式幫我們實現了協程調用方的控制權處理，包括send給協程數據等。我們只要通過async定義協程，await定義阻塞，然后封裝成future的task，放入循環的事件列表中，就等著返回數據。

再來看一個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 協程的應用場景

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

5. 總結

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

•  并發和并行不一樣，并行是同時執行多個任務， 并發是在極短時間內處理多個任務
•  多核cpu，進程是并行，python線程受制于GIL，不能并行，反而因為上下文切換更耗時，協程正好可以彌補
•  協程也不是并行，只是任務交替執行任務，在存在阻塞I/O情況，能夠異步執行，提高效率
•  asyncio 異步I/O庫，可用于開發高并發應用