Python協(xié)程實(shí)踐分享

更新時(shí)間：2022年05月24日 17:00:46 作者：林樹(shù)楷

這篇文章主要分享的是Python協(xié)程實(shí)踐，協(xié)程簡(jiǎn)單來(lái)說(shuō)就是一個(gè)更加輕量級(jí)的線程，并且不由操作系統(tǒng)內(nèi)核管理，完全由程序所控制，下文相關(guān)介紹需要的朋友可以參考一下

協(xié)程

協(xié)程簡(jiǎn)單來(lái)說(shuō)就是一個(gè)更加輕量級(jí)的線程，并且不由操作系統(tǒng)內(nèi)核管理，完全由程序所控制（在用戶(hù)態(tài)執(zhí)行）。協(xié)程在子程序內(nèi)部是可中斷的，然后轉(zhuǎn)而執(zhí)行其他子程序，在適當(dāng)?shù)臅r(shí)候返回過(guò)來(lái)繼續(xù)執(zhí)行。

協(xié)程的優(yōu)勢(shì)？（協(xié)程擁有自己的寄存器上下文和棧，調(diào)度切換時(shí)，寄存器上下文和棧保存到其他地方，在切換回來(lái)的時(shí)候，恢復(fù)先前保存的寄存器上下文和棧，直接操作棧則基本沒(méi)有內(nèi)核切換的開(kāi)銷(xiāo)，可以不加鎖的訪問(wèn)全局變量，所以上下文非常快。）

yield在協(xié)程中的用法

1、協(xié)程中的yield通常出現(xiàn)在表達(dá)式的右邊:

x = yield data

如果yield的右邊沒(méi)有表達(dá)式，默認(rèn)產(chǎn)出的值是None，現(xiàn)在右邊有表達(dá)式，所以返回的是data這個(gè)值。
2、協(xié)程可以從調(diào)用法接受數(shù)據(jù)，調(diào)用通過(guò)send(x)方式將數(shù)據(jù)提供給協(xié)程，同時(shí)send方法中包含next方法，所以程序會(huì)繼續(xù)執(zhí)行。
3、協(xié)程可以中斷執(zhí)行，去執(zhí)行另外的協(xié)程。

經(jīng)典示例

代碼：

def hello():
    data = "mima"
    while True:
        x = yield data  
        print(x)
a = hello()
next(a)
data = a.send("hello")
print(data)

代碼詳解：
程序開(kāi)始執(zhí)行，函數(shù)hello不會(huì)真的執(zhí)行，而是返回一個(gè)生成器給a。
當(dāng)調(diào)用到next()方法時(shí)，hello函數(shù)才開(kāi)始真正執(zhí)行，執(zhí)行print方法，繼續(xù)進(jìn)入while循環(huán)；
程序遇到y(tǒng)ield關(guān)鍵字，程序再次中斷，此時(shí)執(zhí)行到a.send(“hello”)時(shí)，程序會(huì)從yield關(guān)鍵字繼續(xù)向下執(zhí)行，然后又再次進(jìn)入while循環(huán)，再次遇到y(tǒng)ield關(guān)鍵字，程序再次中斷；

協(xié)程在運(yùn)行過(guò)程中的四個(gè)狀態(tài)：

GEN_CREATE:等待開(kāi)始執(zhí)行
GEN_RUNNING:解釋器正在執(zhí)行
GEN_SUSPENDED:在yield表達(dá)式處暫停
GEN_CLOSED:執(zhí)行結(jié)束

生產(chǎn)者-消費(fèi)者模式（協(xié)程）

import time
def consumer():
    r = ""
    while True:
        res = yield r
        if not res:
            print("Starting.....")
            return
        print("[CONSUMER] Consuming %s...." %res)
        time.sleep(1)
        r = "200 OK"
def produce(c):
    next(c)
    n = 0
    while n<6:
        n+=1
        print("[PRODUCER] Producing %s ...."%n)
        r = c.send(n)
        print("[CONSUMER] Consumer return: %s ...."%r)
    c.close()
c = consumer()
produce(c)

代碼分析：

調(diào)用next©啟動(dòng)生成器；
消費(fèi)者一旦生產(chǎn)東西，通過(guò)c.send切換到消費(fèi)者consumer執(zhí)行；
consumer通過(guò)yield關(guān)鍵字獲取到消息，在通過(guò)yield把結(jié)果執(zhí)行；
生產(chǎn)者拿到消費(fèi)者處理過(guò)的結(jié)果，繼續(xù)生成下一條消息；
當(dāng)跳出循環(huán)后，生產(chǎn)者不生產(chǎn)了，通過(guò)close關(guān)閉消費(fèi)者，整個(gè)過(guò)程結(jié)束；

gevent第三方庫(kù)協(xié)程支持

原理：gevent基于協(xié)程的Python網(wǎng)絡(luò)庫(kù)，當(dāng)一個(gè)greenlet遇到IO操作（訪問(wèn)網(wǎng)絡(luò)）自動(dòng)切換到其他的greenlet等到IO操作完成后，在適當(dāng)?shù)臅r(shí)候切換回來(lái)繼續(xù)執(zhí)行。換而言之就是greenlet通過(guò)幫我們自動(dòng)切換協(xié)程，保證有g(shù)reenlet在運(yùn)行，而不是一直等待IO操作。

經(jīng)典代碼

由于切換時(shí)在發(fā)生IO操作時(shí)自動(dòng)完成，所以gevent需要修改Python內(nèi)置庫(kù)，這里可以打上猴子補(bǔ)?。ㄓ脕?lái)在運(yùn)行時(shí)動(dòng)態(tài)修改已有的代碼，而不需要原有的代碼）monkey.patch_all

#!/usr/bin/python2
# coding=utf8
from gevent import monkey
monkey.patch_all()
import gevent
import requests
def handle_html(url):
    print("Starting %s。。。。" % url)
    response = requests.get(url)
    code = response.status_code
    print("%s: %s" % (url, str(code)))
if __name__ == "__main__":
    urls = ["https://www.baidu.com", "https://www.douban.com", "https://www.qq.com"]
    jobs = [ gevent.spawn(handle_html, url) for url in urls ]
    gevent.joinall(jobs)

運(yùn)行結(jié)果：

結(jié)果：3個(gè)網(wǎng)絡(luò)連接并發(fā)執(zhí)行，但是結(jié)束的順序不同。

asyncio內(nèi)置庫(kù)協(xié)程支持

原理:asyncio的編程模型就是一個(gè)消息循環(huán)，從asyncio模塊中直接獲取一個(gè)Eventloop（事件循環(huán)）的應(yīng)用，然后把需要執(zhí)行的協(xié)程放入EventLoop中執(zhí)行，實(shí)現(xiàn)異步IO。

經(jīng)典代碼：

import asyncio
import threading
async def hello():
    print("hello, world: %s"%threading.currentThread())
    await asyncio.sleep(1) # 
    print('hello, man %s'%threading.currentThread())

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait([hello(), hello()]))
    loop.close()

代碼解析：

首先獲取一個(gè)EventLoop
然后將這個(gè)hello的協(xié)程放進(jìn)EventLoop，運(yùn)行EventLoop，它會(huì)運(yùn)行知道future被完成
hello協(xié)程內(nèi)部執(zhí)行await asyncio.sleep(1)模擬耗時(shí)1秒的IO操作，在此期間，主線程并未等待，而是去執(zhí)行EventLoop中的其他線程，實(shí)現(xiàn)并發(fā)執(zhí)行。

代碼結(jié)果：

異步爬蟲(chóng)實(shí)例:

#!/usr/bin/python3
import aiohttp
import asyncio
async def fetch(url, session):
    print("starting: %s" % url)
    async with session.get(url) as response:
        print("%s : %s" % (url,response.status))
        return await response.read()
async def run():
    urls = ["https://www.baidu.com", "https://www.douban.com", "http://www.mi.com"]
    tasks = []
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.ensure_future(fetch(url, session)) for url in urls] # 創(chuàng)建任務(wù)
        response = await asyncio.gather(*tasks) # 并發(fā)執(zhí)行任務(wù)

        for body in response:
            print(len(response))
if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run())
    loop.close()

代碼解析：

創(chuàng)建一個(gè)事件循環(huán)，然后將任務(wù)放到時(shí)間循環(huán)中；
run()方法中主要是創(chuàng)建任務(wù)，并發(fā)執(zhí)行任務(wù)，返回讀取到的網(wǎng)頁(yè)內(nèi)容；
fetch()方法通過(guò)aiohttp發(fā)出指定的請(qǐng)求，以及返回可等待對(duì)象；

（結(jié)束輸出網(wǎng)址和list中網(wǎng)址的順序不同，證明協(xié)程中異步I/O操作）

關(guān)于aiohttp

asyncio實(shí)現(xiàn)類(lèi)TCP、UDP、SSL等協(xié)議，aiohttp則是基于asyncio實(shí)現(xiàn)的HTTP框架，由此可以用來(lái)編寫(xiě)一個(gè)微型的HTTP服務(wù)器。

代碼：

from aiohttp import web
async def index(request):
    await asyncio.sleep(0.5)
    print(request.path)
    return web.Response(body=' Hello, World')
async def hello(request):
    await asyncio.sleep(0.5)
    text = 'hello, %s'%request.match_info['name']
    print(request.path)
    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.0.1:8000....")
    return srv

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(init(loop))
    loop.run_forever()

代碼解析：