Python異步編程_asyncio

1.協程

協程不是計算機提供的，它是程序員創造的。

協程-Co_routine，也可以稱之為微線程，是一種用戶態內的上下文切換技術，簡而言之，其實就是通過一個線程實現代碼塊相互切換執行。

實現協程的幾種方法：

• greenlet，早期模塊

• yield關鍵字

• asyncio裝飾器(python3.4及其之后)

• async、await關鍵字(python3.5及其之后)【推薦】

1.1 greenlet實現協程

pip3 install greenlet

from greenlet import greenlet

def func1():
    print(1)		# step1: output 1
    gr2.switch()	# step3: switch to function func2
    print(2)		# step6: output 2
    gr2.switch()	# step7: 切換到func2函數，從上一次執行的位置繼續向后執行
    
def func2():
    print(3)		# step4: output 3
    gr1.switch()	# step3: 切換到func1函數，從上一次執行的位置繼續向后執行 
    print(4)		# step6: output 4

gr1 = greenlet(func1)
gr2 = greenlet(func2)

gr1.switch()	# step1: 去執行func1函數

1.2 yield關鍵字

def func1():
    yield 1
    yield from func2()
    yield 2

def func2():
    yield 3
    yield 4

f1 = func1()
for item in f1:
    print(item)

1.3 asyncio

在python3.4及其之后的版本支持

import asyncio

@asyncio.corountine
def func1():
    print(1)
    yield from asyncio.sleep(2)	# 遇到io耗時操作，自動化切換到tasks中的其他任務
    print(2)

@asyncio.corountine
def func2():
    print(3)
    yield from asyncio.sleep(2)	# 遇到io耗時操作，自動化切換到tasks中的其他任務
    print(4)

tasks = [
    asyncio.ensure_future(func1()),
    asyncio.ensure_future(func2())
]

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

注意：遇到io阻塞的時候會自動切換，此時線程並沒有閑着，一直在執行任務。

1.4 async & await 關鍵字

在python3.5及其以后版本

import asyncio

async def func1():
    print(1)
    await asyncio.sleep(2)	# 遇到io耗時操作，自動化切換到tasks中的其他任務
    print(2)


async def func2():
    print(3)
    await asyncio.sleep(2)	# 遇到io耗時操作，自動化切換到tasks中的其他任務
    print(4)

tasks = [
    asyncio.ensure_future(func1()),
    asyncio.ensure_future(func2())
]

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

2.協程的意義

充分的利用線程，不讓線程出現阻塞從而浪費硬件資源，同時耗費更少的時間去完成任務。

3.異步編程

3.1 事件循環

一個死循環，去檢測並執行某些代碼7

# 偽代碼：
任務列表 = [任務1, 任務2, 任務3, ...]
while True:
    可執行的任務列表， 已完成的任務列表 = 去任務列表中檢查所有的任務，將‘可執行’和‘已完成’的任務返回
    for 就緒任務 in 可執行任務列表:
        執行已就緒的任務
	for 已完成的任務 in 已完成的任務列表:
        在任務列表中移除 已完成任務
	如果 任務列表中的任務都已完成，則終止循環

import asyncio
# 去生成或獲取一個時間循環
loop = asyncio.get_event_loop()
# 將任務放到'任務列表'
loop.run_until_complete(任務)

3.2 快速上手

首先了解2個定義：

協程函數——定義函數的時候async def 函數名

協程對象——執行協程函數()得到的協程對象

import asyncio

async def func():
    pass
res = func()
# 注意，執行協程函數創建協程對象，函數內部代碼不執行，只是生成了一個協程對象res
# 如果想要運行協程函數的內部代碼，則需要將協程對象res交給事件循環來處理
loop = asyncio.get_event_loop()
loop.run_until_complete(res)
# asyncio.run(res)	# python3.7及其以上版本(可以使用1行代替上面2行代碼)

3.3 await

await + 可等待對象（協程對象、Future對象、Task對象）

示例1：

import asyncio

asyncio def func():
    print("來玩呀！")
	response = await asyncio.sleep(2)
    print("結束", response)

loop = asyncio.get_event_loop()
loop.run_until_complete(func())

示例2：

import asyncio

async def others():
    print("start")
	await asyncio.sleep(2)
    print("end")
    return("others 返回值")

async def func():
    print("協程函數func內部代碼開始執行")
    # 遇到io操作掛起當前協程(任務)，等io操作完成之后再繼續往下執行，當前協程掛起時，事件循環可以去執行其他協程(任務)。
    response = await others()
    print("io 請求結束，結果為：", response)

loop = asyncio.get_event_loop()
loop.run_until_complete(func())

await就是等待對象的值(協程對象對應的函數執行完畢后的返回值)得到之后再繼續往下走

3.4 Task對象

在事件中添加多個任務

Task用於並發調度協程，通過asyncio.create_task(協程對象)得法方式創建Task對象，這樣就可以讓協程加入事件循環中等待被調度執行。除使用asyncio.create_task()函數以外，還可以用低層級的loop.create_task()或asyncio.ensure_future()函數。不建議手動實例化Task對象。

注意：asyncio.create_task()函數在python3.7中被加入，所以在python3.7之前，可以改用低層級的asyncio.ensure_task()函數。

示例 1：

3.5 Future對象

Task繼承Future，Task對象內部await結果的處理是基於Future對象來處理的，它是一種底層的應用，一般不會使用Future而是直接使用Task

import asyncio

async def set_after(fut):
    await asyncio.sleep(2)
    fut.set_result("666")

async def main():
    # 獲取當前 事件循環
    loop = asyncio._get_running_loop()  # python3,7以下
    # loop = asyncio.get_running_loop()  # python3,7及其以上
    # 創建一個任務(Future obj)，沒綁定任何行為，則這個任務會夯住
    fut = loop.create_future()
    # 創建一個任務(Task obj)，綁定了set_after函數，函數內部在2s之后，會給fut的結果賦值。手動設置了fut任務的最終結果，那么fut得到了結果就會結束阻塞，繼續執行下一段代碼。
    await loop.create_task(set_after(fut))
    # 等待Future obj 獲取最終結果，否則會一直阻塞下去
    data = await fut
    print(data)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

3.5.1 擴展：concurrent.futures.Future對象

使用線程池、進程池實現異步操作的時候用到的對象

以后寫代碼可能會存在的交叉時間，例如：crm項目80%都是基於協程異步編程 + 某些模塊(不支持)的時候，那么我們就可以使用進程池或者線程池來做到該某塊的異步編程

def func1():
    # 某個耗時操作
    time.sleep(1)
    return "func1"


async def main():
    loop = asyncio._get_running_loop()
    # loop = asyncio.get_running_loop()  # >=python3.7 required
    # 1. run in the default loop's executor (默認ThreadPoolExecutor)
    # 第一步：內部會先調用ThreadPoolExecutor的submit方法去線程池中申請一個線程去執行func1函數，
    # 並返回一個concurrent.futures.Future對象
    # 第二步：調用asyncio.wrap_future將concurrent.futures.Future對象包裝成asyncio.Future對象。
    # 因為concurrent.futures.Future對象不支持await語法，所以需要包裝為asyncio.Future對象才能使用。
    fut = loop.run_in_executor(None, func1)
    res = await fut
    print("default thread pool =>", res)
    
""""
    # 2. run in a custom thread pool:
    with futures.ThreadPoolExecutor() as pool:
        res = await loop.run_in_executor(pool, func1)
        print("custom thread pool =>", res)
    
    # 3 run in a custom process pool:
    with futures.ProcessPoolExecutor() as pool:
        res = await loop.run_in_executor(pool, func1)
        print("custom process pool =>", res)
"""

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
# asyncio.run(main())  # >=python3.7 required

在使用新的框架時候，有些模塊更新的比較慢，當他不支持asyncio時候就可以使用線程池來實現該模塊的異步編程

案例：asyncio + 不支持異步的模塊

import asyncio
import requests

async def get_image(url):
    # 發送網絡請求，下載圖片（遇到網絡下載io請求，自動化切換到其他下載任務）
    print("開始下載圖片[%s]" % url)

    loop = asyncio.get_event_loop()
    # requests模塊默認不支持異步操作，所以就使用線程池來配合實現異步下載：
    future = loop.run_in_executor(None, requests.get, url)

    response = await future
    print("下載完成")
    # 圖片保存到本地
    file_name = url.rsplit("/")[-1]
    with open(file_name, 'wb') as f:
        f.write(response.content)

if __name__ == '__main__':
    url_list = [
        'https://image11.m1905.cn/uploadfile/2019/0128/20190128095016853342.jpg',
        'https://image11.m1905.cn/uploadfile/2019/0114/20190114103749971987.jpg',
        'https://image11.m1905.cn/uploadfile/2018/1224/20181224014901948964.jpg',
        'https://image11.m1905.cn/uploadfile/2018/0921/20180921031119509766.jpg',
        'https://image11.m1905.cn/uploadfile/2018/1116/20181116095705863204.jpg'
    ]
    tasks = [get_image(url) for url in url_list]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

3.6 異步迭代器

什么是異步迭代器

實現了__aiter__和__anext__方法的對象。__anext__必須返回一個awaitable對象。async for會處理異步迭代器的__anext__方法所返回的可等待對象，知道其引發一個StopAsyncIteration異常。由PEP 492引入。

什么是異步可迭代對象

可在async for語句中被使用的對象。必須通過它的__aiter__方法返回一個asynchronous iterator。由PEP 492引入。

import asyncio

class Reader(object):
    """自定義異步迭代器(也是異步可迭代對象)"""
    def __init__(self):
        self.count = 0

    async def readline(self):
        self.count += 1
        if self.count == 100:
            return None
        return self.count

    def __aiter__(self):
        return self

    async def __anext__(self):
        val = await self.readline()
        if not val:
            raise StopAsyncIteration
        return val

async def main():  # 注意：async for不能單獨使用，必須嵌套在協程函數中使用
    obj = Reader()
    async for item in obj:  # 注意for循環內部實現機制
        print(item)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

.7 異步上下文管理器

此種對象通過定義__aenter__()和__aexit__()方法來對async with語句中的環境進行控制，由PEP492引入。

import asyncio

class AsyncHandleMysql:
    def __init__(self):
        self.conn = None

    async def do_something(self):
        # 這里可以異步操作數據庫
        return 666
    
    async def __aenter__(self):
        # 這里可以異步連接數據庫
        self.conn = await asyncio.sleep(2)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 操作完數據庫之后異步關閉數據庫
        await asyncio.sleep(1)

async def main():
    async with AsyncHandleMysql() as f:  # __aenter__返回什么等於f
        res = await f.do_something()
        print(res)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

4.uvloop

asyncio的事件循環的替代方案，它的事件循環效率是高出python自帶的事件循環效率不少。

使用方法：

安裝：pip install uvloop

使用：

import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())  # 將loop替換成uvloop
# 然后下面的的編寫asyncio代碼不變就可以了
# 內部事件循環會自動變成uvloop
loop = asyncio.get_event_loop()
loop.run_until_complete(...)
# asyncio.run(...)

引申：在Django3.0或者FastApi框架中使用了asgi,而asgi之所以快是使用了uvicorn,其實uvicorn底層就是使用的uvloop作為事件循環

5.實戰案例

5.1 異步操作redis => aioredis

首先，redis要實現異步操作需要安裝模塊：pip install aioredis

然后具體使用示例如下：

import asyncio
import aioredis

async def execute(host, pwd):
    print("開始執行", host)

    # 網絡io操作，先去連接 171.113.244.53:6379，遇到io則自動切換任務，
    # 去連接171.113.244.54:6379
    redis = await aioredis.create_redis_pool(host, password=pwd)
    
    # 網絡io操作，遇到io會自動切換任務
    await redis.hmset_dict("car", key1="BMW", key2="BenZ")
    
    # 網絡io操作，遇到io會自動切換任務
    result = await redis.hgetall("car", encoding="utf-8")
    print(result)
    
    redis.close()
    # 網絡io操作，遇到io自動切換任務
    await redis.wait_closed()
    
    print("結束", host)
    
task_list = [
    execute('redis://171.113.244.53:6379', '12345'),
    execute('redis://171.113.244.54:6379', '67890')
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(task_list))

5.2 異步操作MySQL => aiomysql

首先，MySQL要實現異步操作需要安裝模塊：pip install aiomysql

然后具體使用示例如下：

import asyncio
import aiomysql

async def execute(host, password):
    print("開始", host)
    conn = await aiomysql.connect(
        host=host,
        port=3306,
        user='root',
        password=password,
        db='asyncio_db'
    )
    cur = await conn.cursor()
    await cur.execute("select * from user")
    result = await cur.fetchall()
    print(result)
    
    await cur.close()
    conn.close()
    print("結束", host)
    
task_list = [
    execute('171.113.244.53', '12345'),
    execute('171.113.244.54', '12345')
]

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(task_list))

5.3 FastAPI框架 異步編程

安裝

pip install fastapi
pip install uvicorn

# _*_ coding: utf-8 _*_
# @Date : 2020/5/30 11:25
# @File : test_fastapi_async.py
import asyncio
import uvicorn
import aioredis
from aioredis import Redis
from fastapi import FastAPI

app = FastAPI()
# 創建一個redis連接池
REDIS_POOL = aioredis.ConnectionsPool(
    'redis://127.0.0.1:6379',
    # password='12345',
    minsize=1,
    maxsize=10)

@app.get('/')
def index():
    "普通操作接口"
    return {"message": "Normal Api Index Geted"}

@app.get('/async_test')
async def async_test():
    "異步操作接口"
    print("請求來了")
    await asyncio.sleep(2)
    # 從redis連接池中獲取一個連接
    conn = await REDIS_POOL.acquire()
    redis = Redis(conn)

    # 設置值
    await redis.hmset_dict("book", key1="三國演義", key2="紅樓夢", key3="西游記")
    # 讀取值
    result = await redis.hgetall('book', encoding="utf-8")
    print(result)

    # 歸還連接到redis連接池
    REDIS_POOL.release(conn)
    return result

if __name__ == '__main__':
    uvicorn.run("test_fastapi_async:app",  # test_fastapi_async是代碼所在py文件名
                host='127.0.0.1',
                port=5000,
                log_level='info'
                )

5.4 異步爬蟲 => aiohttp

pip install aiohttp

import aiohttp
import asyncio

async def fetch(session, url):
    print("發送請求:", url)
    async with session.get(url, verify_ssl=False) as response:
        text = await response.text()
        print("得到[%s]結果:" % url, len(text))

async def main():
    async with aiohttp.ClientSession() as session:
        url_list = [
            'https://www.chiphell.com/',
            'http://www.baidu.com',
            'http://www.pythonav.com'
        ]
        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        await asyncio.wait(tasks)

if __name__ == '__main__':
    # asyncio.run(main())
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    
    
# 結果：
發送請求: https://www.chiphell.com/
發送請求: http://www.baidu.com
發送請求: http://www.pythonav.com
得到[http://www.baidu.com]結果: 8074
得到[https://www.chiphell.com/]結果: 5352
得到[http://www.pythonav.com]結果: 3932

## 題外話：
如果你是windows10 + python3.8 + aiohttp3.6.2版本
使用asyncio.run(main())運行時候會報錯：
“RuntimeError: Event loop is closed”
該錯誤可能是aiohttp與python3.8的兼容性bug。

總結

異步編程的主要目的在於：通過一個線程利用其io等待時間達到用更少的資源去做更多的事的目的