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等待時間達到用更少的資源去做更多的事的目的

請相信自己

當我們迷茫，懶惰，退縮的時候

我們會格外的相信命運

相信一切都是命中注定

而當我們努力拼搏，積極向上時

我們會格外的相信自己

是以命運是什麼呢？

它是如果你習慣它

那它就會一直左右你

如果你想掙脫它

那它就成為你的阻礙

可如果你打破了它

那它就是你人生的墊腳石!

如果覺得這篇文章對你有小小的幫助的話，記得在右下角點個“推薦”哦，部落客在此感謝！