python協程(yield、asyncio標准庫、gevent第三方)、異步的實現

引言

同步：不同程序單元為了完成某個任務，在執行過程中需靠某種通信方式以協調一致，稱這些程序單元是同步執行的。

例如購物系統中更新商品庫存，需要用“行鎖”作為通信信號，讓不同的更新請求強制排隊順序執行，那更新庫存的操作是同步的。

簡言之，同步意味着有序。

阻塞：程序未得到所需計算資源時被掛起的狀態。

程序在等待某個操作完成期間，自身無法繼續干別的事情，則稱該程序在該操作上是阻塞的。

常見的阻塞形式有：網絡I/O阻塞、磁盤I/O阻塞、用戶輸入阻塞等。

阻塞狀態下的性能提升

引入多進程：

在一個程序內，依次執行10次太耗時，那開10個一樣的程序同時執行不就行了。於是我們想到了多進程編程。為什么會先想到多進程呢？發展脈絡如此。在更早的操作系統（Linux 2.4）及其以前，進程是 OS 調度任務的實體，是面向進程設計的OS.

改善效果立竿見影。但仍然有問題。總體耗時並沒有縮減到原來的十分之一，而是九分之一左右，還有一些時間耗到哪里去了？進程切換開銷。

進程切換開銷不止像“CPU的時間觀”所列的“上下文切換”那么低。CPU從一個進程切換到另一個進程，需要把舊進程運行時的寄存器狀態、內存狀態全部保存好，再將另一個進程之前保存的數據恢復。對CPU來講，幾個小時就干等着。當進程數量大於CPU核心數量時，進程切換是必然需要的。
除了切換開銷，多進程還有另外的缺點。一般的服務器在能夠穩定運行的前提下，可以同時處理的進程數在數十個到數百個規模。如果進程數量規模更大，系統運行將不穩定，而且可用內存資源往往也會不足。
多進程解決方案在面臨每天需要成百上千萬次下載任務的爬蟲系統，或者需要同時搞定數萬並發的電商系統來說，並不適合。
除了切換開銷大，以及可支持的任務規模小之外，多進程還有其他缺點，如狀態共享等問題，后文會有提及，此處不再細究。

多線程(改進(多進程帶來多問題))

由於線程的數據結構比進程更輕量級，同一個進程可以容納多個線程，從進程到線程的優化由此展開。后來的OS也把調度單位由進程轉為線程，進程只作為線程的容器，用於管理進程所需的資源。而且OS級別的線程是可以被分配到不同的CPU核心同時運行的。

結果符合預期，比多進程耗時要少些。從運行時間上看，多線程似乎已經解決了切換開銷大的問題。而且可支持的任務數量規模，也變成了數百個到數千個。

但是，多線程仍有問題，特別是Python里的多線程。首先，Python中的多線程因為GIL的存在，它們並不能利用CPU多核優勢，一個Python進程中，只允許有一個線程處於運行狀態。那為什么結果還是如預期，耗時縮減到了十分之一？

因為在做阻塞的系統調用時，例如sock.connect(),sock.recv()時，當前線程會釋放GIL，讓別的線程有執行機會。但是單個線程內，在阻塞調用上還是阻塞的。

小提示：Python中 time.sleep 是阻塞的，都知道使用它要謹慎，但在多線程編程中，time.sleep 並不會阻塞其他線程。

除了GIL之外，所有的多線程還有通病。它們是被OS調度，調度策略是搶占式的，以保證同等優先級的線程都有均等的執行機會，那帶來的問題是：並不知道下一時刻是哪個線程被運行，也不知道它正要執行的代碼是什么。所以就可能存在競態條件。

例如爬蟲工作線程從任務隊列拿待抓取URL的時候，如果多個爬蟲線程同時來取，那這個任務到底該給誰？那就需要用到“鎖”或“同步隊列”來保證下載任務不會被重復執行。

而且線程支持的多任務規模，在數百到數千的數量規模。在大規模的高頻網絡交互系統中，仍然有些吃力。當然，多線程最主要的問題還是競態條件。

非阻塞(阻塞狀態下的性能提升)

原非阻方案

def nonblocking_way():
    sock = socket.socket()
    sock.setblocking(False)
    try:
        sock.connect(('example.com', 80))
    except BlockingIOError:
        # 非阻塞連接過程中也會拋出異常
        pass
    request = 'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n'
    data = request.encode('ascii')
    # 不知道socket何時就緒，所以不斷嘗試發送
    while True:
        try:
            sock.send(data)
            # 直到send不拋異常，則發送完成
            break
        except OSError:
            pass

    response = b''
    while True:
        try:
            chunk = sock.recv(4096)
            while chunk:
                response += chunk
                chunk = sock.recv(4096)
            break
        except OSError:
            pass
    return response

首先注意到兩點，就感覺被騙了。一是耗時與同步阻塞相當，二是代碼更復雜。要非阻塞何用？且慢。
上第9行代碼sock.setblocking(False)告訴OS，讓socket上阻塞調用都改為非阻塞的方式。之前我們說到，非阻塞就是在做一件事的時候，不阻礙調用它的程序做別的事情。上述代碼在執行完 sock.connect() 和 sock.recv() 后的確不再阻塞，可以繼續往下執行請求准備的代碼或者是執行下一次讀取。

代碼變得更復雜也是上述原因所致。第11行要放在try語句內，是因為socket在發送非阻塞連接請求過程中，系統底層也會拋出異常。connect()被調用之后，立即可以往下執行第15和16行的代碼。

需要while循環不斷嘗試 send()，是因為connect()已經非阻塞，在send()之時並不知道 socket 的連接是否就緒，只有不斷嘗試，嘗試成功為止，即發送數據成功了。recv()調用也是同理。
雖然 connect() 和 recv() 不再阻塞主程序，空出來的時間段CPU沒有空閑着，但並沒有利用好這空閑去做其他有意義的事情，而是在循環嘗試讀寫 socket （不停判斷非阻塞調用的狀態是否就緒）。還得處理來自底層的可忽略的異常。也不能同時處理多個 socket 。

非阻塞改進

1. epoll
   判斷非阻塞調用是否就緒如果 OS 能做，是不是應用程序就可以不用自己去等待和判斷了，就可以利用這個空閑去做其他事情以提高效率。
   所以OS將I/O狀態的變化都封裝成了事件，如可讀事件、可寫事件。並且提供了專門的系統模塊讓應用程序可以接收事件通知。這個模塊就是select。讓應用程序可以通過select注冊文件描述符和回調函數。當文件描述符的狀態發生變化時，select 就調用事先注冊的回調函數。
   select因其算法效率比較低，后來改進成了poll，再后來又有進一步改進，BSD內核改進成了kqueue模塊，而Linux內核改進成了epoll模塊。這四個模塊的作用都相同，暴露給程序員使用的API也幾乎一致，區別在於kqueue 和 epoll 在處理大量文件描述符時效率更高。
   鑒於 Linux 服務器的普遍性，以及為了追求更高效率，所以我們常常聽聞被探討的模塊都是 epoll

2 . 回調((Callback))
把I/O事件的等待和監聽任務交給了 OS，那 OS 在知道I/O狀態發生改變后（例如socket連接已建立成功可發送數據），它又怎么知道接下來該干嘛呢？只能回調。
需要我們將發送數據與讀取數據封裝成獨立的函數，讓epoll代替應用程序監聽socket狀態時，得告訴epoll：“如果socket狀態變為可以往里寫數據（連接建立成功了），請調用HTTP請求發送函數。如果socket 變為可以讀數據了（客戶端已收到響應），請調用響應處理函數。”
首先，不斷嘗試send() 和 recv() 的兩個循環被消滅掉了。
其次，導入了selectors模塊，並創建了一個DefaultSelector 實例。Python標准庫提供的selectors模塊是對底層select/poll/epoll/kqueue的封裝。DefaultSelector類會根據 OS 環境自動選擇最佳的模塊，那在 Linux 2.5.44 及更新的版本上都是epoll了。
然后，在第25行和第31行分別注冊了socket可寫事件(EVENT_WRITE)和可讀事件(EVENT_READ)發生后應該采取的回調函數。

雖然代碼結構清晰了，阻塞操作也交給OS去等待和通知了，但是，我們要抓取10個不同頁面，就得創建10個Crawler實例，就有20個事件將要發生，那如何從selector里獲取當前正發生的事件，並且得到對應的回調函數去執行呢？

3. 事件循環（Event Loop）
   為了解決上述問題，那我們只得采用老辦法，寫一個循環，去訪問selector模塊，等待它告訴我們當前是哪個事件發生了，應該對應哪個回調。這個等待事件通知的循環，稱之為事件循環。
   重要的是第49行代碼，selector.select() 是一個阻塞調用，因為如果事件不發生，那應用程序就沒事件可處理，所以就干脆阻塞在這里等待事件發生。那可以推斷，如果只下載一篇網頁，一定要connect()之后才能send()繼而recv()，那它的效率和阻塞的方式是一樣的。因為不在connect()/recv()上阻塞，也得在select()上阻塞。
   '
   python3 asyncio封裝上述非阻塞的改進方法，用時4年打造異步標准庫

協程引入

yield

def consumer():
    r = ''
    while True:
        n = yield r
        if not n:
            return
        print('[CONSUMER] Consuming %s...' % n)
        r = '200 OK'

def produce(c):
    c.send(None)
    n = 0
    while n < 5:
        n = n + 1
        print('[PRODUCER] Producing %s...' % n)
        r = c.send(n)
        print('[PRODUCER] Consumer return: %s' % r)
    c.close()

c = consumer()
produce(c)

demo解析：
注意到consumer函數是一個generator，把一個consumer傳入produce后：
首先調用c.send(None)啟動生成器；
然后，一旦生產了東西，通過c.send(n)切換到consumer執行；
consumer通過yield拿到消息，處理，又通過yield把結果傳回；
produce拿到consumer處理的結果，繼續生產下一條消息；
produce決定不生產了，通過c.close()關閉consumer，整個過程結束

在 Python 中調用協程對象1的 send() 方法時，第一次調用必須使用參數 None, 這使得協程的使用變得十分麻煩
解決此問題：
借助 Python 自身的特性來避免這一問題，比如，創建一個裝飾器

def routine(func):
    def start(*args, **kwargs):
        cr = func(*args, **kwargs)
        cr.send(None)
        return cr
    return start
		
@routine
def product():
	pass

yield from

yield from 是Python 3.3 新引入的語法（PEP 380）。它主要解決的就是在生成器里玩生成器不方便的問題。它有兩大主要功能。
第一個功能是：讓嵌套生成器不必通過循環迭代yield，而是直接yield from。以下兩種在生成器里玩子生成器的方式是等價的。

def gen_one():

    subgen = range(10)    yield from subgendef gen_two():

    subgen = range(10)    for item in subgen:        yield item

第二個功能就是在子生成器和原生成器的調用者之間打開雙向通道，兩者可以直接通信。

def gen():
    yield from subgen()def subgen():
    while True:
        x = yield
        yield x+1def main():
    g = gen()
    next(g)                # 驅動生成器g開始執行到第一個 yield
    retval = g.send(1)     # 看似向生成器 gen() 發送數據
    print(retval)          # 返回2
    g.throw(StopIteration) # 看似向gen()拋入異常

用yield from改進基於生成器的協程，代碼抽象程度更高。使業務邏輯相關的代碼更精簡。由於其雙向通道功能可以讓協程之間隨心所欲傳遞數據，使Python異步編程的協程解決方案大大向前邁進了一步。
於是Python語言開發者們充分利用yield from，使 Guido 主導的Python異步編程框架Tulip迅速脫胎換骨，並迫不及待得讓它在 Python 3.4 中換了個名字asyncio以“實習生”角色出現在標准庫中。

asyncio

asyncio是Python 3.4 試驗性引入的異步I/O框架（PEP 3156），提供了基於協程做異步I/O編寫單線程並發代碼的基礎設施。其核心組件有事件循環（Event Loop）、協程(Coroutine）、任務(Task)、未來對象(Future)以及其他一些擴充和輔助性質的模塊。

在引入asyncio的時候，還提供了一個裝飾器@asyncio.coroutine用於裝飾使用了yield from的函數，以標記其為協程。但並不強制使用這個裝飾器。

• 協程的隱式創建方式

import threading
import asyncio

@asyncio.coroutine
def hello():
    print('Hello world! (%s)' % threading.currentThread())
    yield from asyncio.sleep(1)
    print('Hello again! (%s)' % threading.currentThread())

loop = asyncio.get_event_loop()
tasks = [hello(), hello()]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

@asyncio.coroutine把一個generator標記為coroutine類型，然后，我們就把這個coroutine扔到EventLoop中執行。
hello()會首先打印出Hello world!，然后，yield from語法可以讓我們方便地調用另一個generator。由於asyncio.sleep()也是一個coroutine，所以線程不會等待asyncio.sleep()，而是直接中斷並執行下一個消息循環。當asyncio.sleep()返回時，線程就可以從yield from拿到返回值（此處是None），然后接着執行下一行語句。
把asyncio.sleep(1)看成是一個耗時1秒的IO操作，在此期間，主線程並未等待，而是去執行EventLoop中其他可以執行的coroutine了，因此可以實現並發執行。

• 協程顯示的創建方式(便於理解隱式創建)
  創建任務task(2種)

import asyncio
import time
now = lambda : time.time()
async def do_some_work(x):
    print('Waiting: ', x)
start = now()
coroutine = do_some_work(2)
loop = asyncio.get_event_loop()
# task = asyncio.ensure_future(coroutine)    # 方式一
task = loop.create_task(coroutine)    # 方式二
print(task)
loop.run_until_complete(task)
print(task)
print('TIME: ', now() - start)

創建task后，task在加入事件循環之前是pending狀態，加入loop后運行中是running狀態，loop調用完是Done，運行完是finished狀態，雖說本質上協程函數和task指的東西都一樣，但是task有了協程函數的狀態。
其中loop.run_until_complete()接受一個future參數，futurn具體指代一個協程函數，而task是future的子類，所以我們不聲明一個task直接傳入協程函數也能執行。

• 協程綁定綁定回調函數
  通過task的task.add_done_callback(callback)方法綁定回調函數，回調函數接收一個future對象參數如task，在內部通過future.result()獲得協程函數的返回值。

import asyncio
async def test(x):
    return x+3
def callback(y):
    print(y.result())
coroutine = test(5)
loop = asyncio.get_event_loop()
task = loop.create_task(coroutine)
task
<Task pending coro=<test() running at <ipython-input-4-61142fef17d8>:1>>
task.add_done_callback(callback)
loop.run_until_complete(task)

• 協程耗時任務掛起
  多任務聲明了協程函數，也同時在loop中注冊了，他的執行也是順序執行的，因為在異步函數中沒有聲明那些操作是耗時操作，所以會順序執行。await的作用就是告訴控制器這個步驟是耗時的，async可以定義協程對象，使用await可以針對耗時的操作進行掛起

import asyncio
import time
async def test(1):
    time.sleep(1)
    print(time.time())
tasks = [asyncio.ensure_future(test()) for _ in range(3)]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

1547187398.7611663
1547187399.7611988
1547187400.7632194

上面執行並不是異步執行，而是順序執行，但是改成下面形式那就是異步執行：

import asyncio
import time
async def test(t):
    await asyncio.sleep(1)
    print(time.time())
tasks = [asyncio.ensure_future(test()) for _ in range(3)]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

1547187398.7611663
1547187399.7611988
1547187400.7632194

async、await(將耗時函數掛起)

用asyncio提供的@asyncio.coroutine可以把一個generator標記為coroutine類型，然后在coroutine內部用yield from調用另一個coroutine實現異步操作。
為了簡化並更好地標識異步IO，從Python 3.5開始引入了新的語法async和await，可以讓coroutine的代碼更簡潔易讀。
請注意，async和await是針對coroutine的新語法，要使用新的語法，只需要做兩步簡單的替換：

把@asyncio.coroutine替換為async；
把yield from替換為await。

async def hello():
    print("Hello world!")
    r = await asyncio.sleep(1)
    print("Hello again!")

aiohttp

asyncio可以實現單線程並發IO操作。如果僅用在客戶端，發揮的威力不大。如果把asyncio用在服務器端，例如Web服務器，由於HTTP連接就是IO操作，因此可以用單線程+coroutine實現多用戶的高並發支持。

asyncio實現了TCP、UDP、SSL等協議，aiohttp則是基於asyncio實現的HTTP框架。

import asyncio

from aiohttp import web

async def index(request):
    await asyncio.sleep(0.5)
    return web.Response(body=b'<h1>Index</h1>')

async def hello(request):
    await asyncio.sleep(0.5)
    text = '<h1>hello, %s!</h1>' % request.match_info['name']
    return web.Response(body=text.encode('utf-8'))

async def init(loop):
    app = web.Application(loop=loop)
    app.router.add_route('GET', '/', index)
    app.router.add_route('GET', '/hello/{name}', hello)
    srv = await loop.create_server(app.make_handler(), '127.0.0.1', 8000)
    print('Server started at http://127.0.0.1:8000...')
    return srv

loop = asyncio.get_event_loop()                # 創建一個事件循環（池）
loop.run_until_complete(init(loop))           # 將協程對象包裝並注冊協程對象
loop.run_forever()

多進程配合使用 + 協程

方法1:
asyncio、aiohttp需要配合aiomultiprocess

方法2:
gevent.pool import Pool
multiprocessing import Process
核心代碼

def main():
        file_list = ["7001", "7002", "7003"]

        p_lst = []  # 線程列表
        for i in file_list:
            # self.run(i)
            p = Process(target=read_file, args=(i,))  # 子進程調用函數
            p.start()  # 啟動子進程
            p_lst.append(p)  # 將所有進程寫入列表中

def read_file(self, number):
        """
        讀取文件
        :param number: 文件標記
        :return:
        """
        file_name = os.path.join(self.BASE_DIR, "data", "%s.txt" % number)
        # print(file_name)
        self.write_log(number, "開始讀取文件 {}".format(file_name),"green")
        with open(file_name, encoding='utf-8') as f:
            # 使用協程池，執行任務。語法: pool.map(func,iterator)
            # partial使用偏函數傳遞參數
            # 注意：has_null第一個參數,必須是迭代器遍歷的值
            pool.map(partial(self.has_null, number=number), f)

多協程

使用loop.run_until_complete(syncio.wait(tasks)) 也可以使用 loop.run_until_complete(asyncio.gather(*tasks)） ,前者傳入task列表，會對task進行解包操作。

多協程嵌套

async def get(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            print(response)
            print(time.time())

import time
async def request():
    url = "http://www.baidu.com"
    resulit = await get(url)

tasks = [asyncio.ensure_future(request()) for _ in range(10000)]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

• 返回方式(3種)

async def get(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            print(response)
            print(time.time())
async def request():
    url = "http://www.baidu.com"
    tasks = [asyncio.ensure_future(url) for _ in range（1000）]
方式一：
    dones, pendings = await asyncio.wait(tasks) # 返回future對象，不返回直接結果
    for task in dones:
        print('Task ret: ', task.result())
方式二：
    results = await asyncio.gather(*tasks) # 直接返回結果

方式三：
    for task in asyncio.as_completed(tasks):
        result = await task
        print('Task ret: {}'.format(result)) # 迭代方式返回結果

tasks = asyncio.ensure_future(request())
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

停止協程任務

實現結束task有兩種方式：關閉單個task、關閉loop，涉及主要函數：

asyncio.Task.all_tasks()獲取事件循環任務列表

KeyboardInterrupt捕獲停止異常（Ctrl+C）

loop.stop()停止任務循環

task.cancel()取消單個任務

loop.run_forever()

loop.close()關閉事件循環，不然會重啟

gevent(第三方包實現協程方式)

python程序實現的一種單線程下的多任務執行調度器，簡單來說在一個線程里，先后執行AB兩個任務，但是當A遇到耗時操作（網絡等待、文件讀寫等），這個時候gevent會讓A繼續執行，但是同時也會開始執行B任務，如果B在遇到耗時操作同時A又執行完了耗時操作，gevent又繼續執行A。

import gevent
def test(time):
    print(1)
    gevent.sleep(time)
    print(2)
def test2(time):
    print(3)
    gevent.sleep(time)
    print(4)
if __name__ == '__main__':
    gevent.joinall([
        gevent.spawn(test, 2),
        gevent.spawn(test2, 3)
    ])

借鑒文章：
https://mp.weixin.qq.com/s/GgamzHPyZuSg45LoJKsofA
https://rgb-24bit.github.io/blog/2019/python-coroutine-event-loop.html
https://zhuanlan.zhihu.com/p/54657754
https://cloud.tencent.com/developer/article/1590280