Python協程&asyncio&異步編程

1.協程

協程是微執行緒，是一種用戶態背景關系切換技術，通過一個執行緒實作代碼塊相互切換執行
0BADD6AB28BA49E1A9AEC2A831972844
實作協程有這么幾種方法：

greenlet，早期的模塊
yield 關鍵字
asyncio python3.4引入的
async、await關鍵字 python3.5 主流[推薦]

1.1 greenlet實作協程

pip install greenlet

# -*- coding: utf-8 -*-
from greenlet import greenlet


def func1():
    print(1)       # 第1步：輸出1
    gr2.switch()   # 第2步：跳到func2函式
    print(2)       # 第5步：輸出2
    gr2.switch()   # 第6步：跳到func2函式


def func2():
    print(3)      # 第3步：輸出3
    gr1.switch()  # 第4步：跳到func1函式
    print(4)      # 第7步：輸出4


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

gr1.switch()   # 第1步：去執行func1函式

1.2 yield關鍵字

# -*- coding: utf-8 -*-


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以及之后

# -*- coding: utf-8 -*-
import asyncio


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


@asyncio.coroutine
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及以后

# -*- coding: utf-8 -*-
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))

1.5 一般用greenlet或者async&await關鍵字

2.協程的意義

在一個執行緒中遇到IO等待時間，不讓執行緒一直白白等待，而是讓執行緒利用空閑時間去點其它的事

# -*- coding: utf-8 -*-
import asyncio
import aiohttp


async def fetch(session, url):
    print("發送請求")
    async with session.get(url, verify_ssl=False) as response:
        content = await response.content.read()
        file_name = url.rsplit("/")[-1]
        with open(file_name, mode="wb") as file_object:
            file_object.write(content)
        print("下載完成", url)


async def main():
    async with aiohttp.ClientSession() as session:
        url_list = [
            "http://ww1.sinaimg.cn/mw600/00745YaMgy1gedxxa59lyj30kk10p77m.jpg",
            "http://ww1.sinaimg.cn/mw600/00745YaMgy1gedxrlhlhaj30kk0dpmxj.jpg",
            "http://ww1.sinaimg.cn/mw600/00745YaMgy1gedxrlrw4tj30kk0pp78u.jpg"
        ]

        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        await asyncio.wait(tasks)

if __name__ == "__main__":
    asyncio.run(main())

3.異步編程

3.1 事件回圈

理解成為一個死回圈

"""
# 偽代碼

任務串列 = [任務1， 任務2， 任務3，...]

while True:
    可執行的任務串列，已完成的任務串列 = 去任務串列中檢查所有任務，將'可執行'和'已完成'的任務回傳
    for 就緒任務 in 可執行的任務串列:
        執行就緒任務
    for 已完成任務 in 已完成的任務串列:
        在任務串列中 移除已完成任務
    任務串列中的任務都已完成 則終止回圈
"""

import asyncio

# 去生成或獲取一個事件回圈
loop = asyncio.get_event_loop()
# 把任務放到任務串列
loop.run_until_complete(asyncio.wait(tasks))

3.2 快速上手

協程函式定義函式的時候 async def 函式名
協程物件執行協程函式()得到協程物件

async def func():
    pass
    
result = func()

注意：運行協程函式得到協程物件，函式內部代碼不會執行
要運行協程函式內部代碼，必須要將協程物件交給事件回圈來處理

import asyncio

async def func():
	print("Hello World")
    
result = func()

# 去生成或獲取一個事件回圈
loop = asyncio.get_event_loop()
# 把任務放到任務串列
loop.run_until_complete(result)
# asyncio.run(result) Python3.7之后，這句可以代替上面兩句

3.3 await

await + 可等待物件(協程物件、Future、Task物件、-->IO等待)
示例1:

import asyncio


async def func():
    print("start")
    response = await asyncio.sleep(2)
    print("end", response)
    
asyncio.run(func)

示例2:

# -*- coding: utf-8 -*-
import asyncio


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


async def func():
    print("開始執行func")
    response = await others()
    print("IO請求結束，結果為: ", response)


asyncio.run(func())

示例3:

# -*- coding: utf-8 -*-
import asyncio


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


async def func():
    print("開始執行func")
    response1 = await others()
    print("IO請求結束，結果為: ", response1)

    response2 = await others()
    print("IO請求結束，結果為: ", response2)

asyncio.run(func())

await就是等待物件的值得到對應的結果之后再繼續往下走

3.4 Task物件

在事件回圈中添加多個任務
Tasks用語并發調度協程，通過asyncio.create_task(協程物件)，這樣可以讓協程加入事件回圈中等待被調度執行，除了使用asyncio.create_task(協程物件)還可以使用較低級的loop.create_task()或ensure_future()函式，不建議手動實體化Task物件，
注意：asyncio.create_task(協程物件)是在Python3.7時加入的，在Python3.7前可以使用較低級的asyncio.ensure_future()函式，
示例1:

import asyncio


async def func():
    print("1")
    await asyncio.sleep(2)
    print("2")
    return "回傳值"


async def main():
    print("開始執行main")
    # 創建Task物件，將當前執行func函式任務添加到事件回圈
    task1 = asyncio.create_task(func())
    task2 = asyncio.create_task(func())
    print("main結束")

    # 當執行協程遇到IO操作時，會自動化切換到其它任務(task2)
    # 此處的await是等待相對應的協程全部執行完畢并獲取結果
    ret1 = await task1
    ret2 = await task2

    print(ret1, ret2)

asyncio.run(main())

示例2:

# -*- coding: utf-8 -*-
import asyncio


async def func():
    print("1")
    await asyncio.sleep(2)
    print("2")
    return "回傳值"


async def main():
    print("開始執行main")
    # 創建Task物件，將當前執行func函式任務添加到事件回圈
    task_list = [
        asyncio.create_task(func(), name="n1"),
        asyncio.create_task(func(), name="n2")
    ]

    print("main結束")

    # 當執行協程遇到IO操作時，會自動化切換到其它任務(task2)
    # 此處的await是等待相對應的協程全部執行完畢并獲取結果
    # done是集合，是上面兩個任務的回傳值
    done, pending = await asyncio.wait(task_list, timeout=None)

    print(done)

asyncio.run(main())

示例3:

import asyncio


async def func():
    print("1")
    await asyncio.sleep(2)
    print("2")
    return "回傳值"

task_list = [
    func(),
    func()
]

done, pending = asyncio.run(asyncio.wait(task_list))
print(done)

3.5 asyncio.Future物件

Task繼承Future，Task物件內部的await結果處理基于Future物件來著的
示例1:

import asyncio


async def main():
    # 獲取當前事件回圈
    loop = asyncio.get_running_loop()
    # 創建一個future物件，這個物件什么都不干
    fut = asyncio.create_future()
    # 等待任務最終結果(Future物件)，沒有結果則會一直等下去
    await fut


asyncio.run(main())

示例2:

import asyncio


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

async def main():
    # 獲取當前事件回圈
    loop = asyncio.get_running_loop()
    # 創建一個future物件，這個物件什么都不干，沒系結任何行為，則這個任務不知道什么時候結束
    fut = asyncio.create_future()

    # 創建一個任務(Task物件)，系結了set_after函式，函式在2s之后，會給fut賦值
    # 即手動給fut設定結果，那么fut就可以結束了
    await loop.create_task(set_after(fut))

    # 等待任務最終結果(Future物件)，沒有結果則會一直等下去
    data = https://www.cnblogs.com/guotianbao/p/await fut
    print(data)


asyncio.run(main())

3.6 concurrent.future.Future物件

使用執行緒池或行程池實作異步操作時用到的物件
交叉使用，協程異步編程 + MySQL(不支持)[執行緒、行程做異步編程]

import time
import asyncio
import concurrent.futures


def func():
    # 某個耗時操作
    time.sleep(2)
    return "success"


async def main():
    loop = asyncio.get_running_loop()

    # 1.Run in the default loop's excutor (默認ThreadPoolExecutor)
    # 第一步：內部會先呼叫ThreadPoolExecutor的submit方法去執行緒池中申請一個執行緒去執行func函式，并回傳一個concurrent.futures.Future物件
    # 第二步：呼叫asyncio.wrap_future將concurrent.futures.Future物件包裝為呼叫asyncio.Future物件
    # 因為concurrent.futures.Future物件不支持await語法，需要包裝為asyncio.Future物件才能使用
    fut = loop.run_in_excutor(None, func)
    result = await fut
    print("default thread pool", result)

    # 2.Run in a custom thread pool
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_excutor(pool, func)
        print("custom thread pool", result)

    # 3.Run in a custom process pool
    with concurrent.futures.ProcessPoolExecutor() as pool:
        result = await loop.run_in_excutor(pool, func)
        print("custom process pool", result)


asyncio.run(main())

案例：asyncio + 不支持的模塊

import asyncio
import requests


async def download_image(url):
    # f發送網路請求下載圖片，(遇到下載圖片的網路IO請求，自動切換到其它任務)
    print("開始下載", url)
    loop = asyncio.get_running_loop()

    # requests模塊不支持異步操作，所以就使用執行緒池來配合實作了
    future = loop.run_in_executor(None, requests.get, url)

    response = await future
    print("下載完成")

    # 圖片保存到本地檔案
    file_name = url.rsplit("/")[-1]
    with open(file_name, mode="wb") as file_object:
        file_object.write(response.content)


def main():

    url_list = [
        "http://ww1.sinaimg.cn/mw600/00745YaMgy1gedxxa59lyj30kk10p77m.jpg",
        "http://ww1.sinaimg.cn/mw600/00745YaMgy1gedxrlhlhaj30kk0dpmxj.jpg",
        "http://ww1.sinaimg.cn/mw600/00745YaMgy1gedxrlrw4tj30kk0pp78u.jpg"
    ]

    tasks = [download_image(url) for url in url_list]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))


if __name__ == "__main__":
    main()

3.7 異步迭代器

什么是異步迭代器
實作了__aiter__()和__anext__()方法的物件，anext__必須回傳一個awaitable物件，async for 會處理異步迭代器的__anext()方法所回傳的可等待物件，知道其引發一個StopAsyncIteration例外，
什么是異步可迭代物件
可在async for 中使用的物件，必須通過它的__aiter__()回傳一個asynchronous iterator

import asyncio


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

    async def readline(self):
        # await asyncio.sleep(2)
        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 val == None:
            raise StopAsyncIteration
        return val


async def func():
    obj = Reader()
    # async for只能寫在協程函式內
    async for item in obj:
        print(item)

asyncio.run(func())

3.8 異步背景關系管理器

此種物件通過定義__aenter__()方法和__aexit__()方法來對async with陳述句中的環境進行控制

import asyncio


class AsyncContexManager(object):
    def __init__(self, ):
        self.conn = conn

    async def do_something(self):
        # 異步操作資料庫
        return 666

    async def __aenter__(self):
        # 異步連接資料庫
        self.conn = await asyncio.sleep(2)
        return self

    async def __anext__(self):
        # 異步關閉資料庫
        await asyncio.sleep(2)


async def func():
    # async with 只能在協程函式中使用
    async with AsyncContexManager() as f:
        result = await f.do_something()
        print(result)

asyncio.run(func())

4.uvloop

是asyncio的事件回圈的替代方案，事件回圈 > asyncio的事件回圈，性能比肩go

pip install uvloop

import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy)


#  撰寫asyncio的代碼，與之前寫的代碼一致

# 內部的事件回圈會自動變化為uvloop

asyncio.run(...)

注意：一個asgi-->uvicorn 內部使用的是uvloop

5.實戰案例

5.1 異步操作redis

在使用Python代碼操作redis，連接、操作、斷開都是網路IO

pip install aioredis

示例1:

import asyncio
import aioredis


async def execute(address, password):
    print("開始執行", address)
    # 網路IO 創建redis連接
    redis = await aioredis.create_redis(address, password=password)
    # 網路IO 在redis中設定哈希值car，內部設定3個鍵值對 redis = {"car": {"key1": 1, "key2": 2, "key3": 3}}
    await redis.hmset_dict("car", key1=1, key2=2, key3=3)
    # 網路IO 去redis中獲取值
    result = redis.hgetall("car", encoding="utf-8")
    print(result)

    redis.close()
    # 網路IO關閉redis連接
    await redis.wait_close()

    print("結束", address)


asyncio.run(execute("redis://127.0.0.1:6379", "123456"))

示例2:

import asyncio
import aioredis


async def execute(address, password):
    print("開始執行", address)
    # 網路IO 先去連接47.93.4.197:6379，遇到IO則自動切換任務，去連接47.93.4.198:6379
    redis = await aioredis.create_redis(address, password=password)
    # 網路IO 遇到IO會自動切換任務
    await redis.hmset_dict("car", key1=1, key2=2, key3=3)
    # 網路IO 遇到IO會自動切換任務
    result = redis.hgetall("car", encoding="utf-8")
    print(result)

    redis.close()
    # 網路IO 遇到IO會自動切換任務
    await redis.wait_close()

    print("結束", address)


task_list = [
    execute("47.93.4.197:6379", "123456"),
    execute("47.93.4.198:6379", "123456"),
]

asyncio.run(asyncio.wait(task_list))

5.1 異步操作MySQL

pip install aiomysql

示例1:

import asyncio
import aiomysql


async def execute():
    print("開始執行")
    # 網路IO 連接MySQL
    conn = await aiomysql.connect(host="127.0.0.1", port=3306, user="root", password="123456", db="mysql")

    # 網路IO 創建cursor
    cur = await conn.cursor()
    # 網路IO 執行sql
    await cur.execute("seletc name from user")
    # 網路IO 獲取sql結果
    result = await cur.fetchall()
    print(result)
    # 網路IO 關閉連接
    await cur.close()
    conn.close()


asyncio.run(execute())

示例2:

import asyncio
import aiomysql


async def execute(host, password):
    print("開始執行", host)
    # 網路IO 連接MySQL先去連接47.93.41.197，遇到IO則切換去連接47.93.41.198
    conn = await aiomysql.connect(host=host, port=3306, user="root", password=password, db="mysql")

    # 網路IO 遇到IO會自動切換任務
    cur = await conn.cursor()
    # 網路IO 遇到IO會自動切換任務
    await cur.execute("seletc name from user")
    # 網路IO 遇到IO會自動切換任務
    result = await cur.fetchall()
    print(result)
    # 網路IO 遇到IO會自動切換任務
    await cur.close()
    conn.close()
    print("結束", host)


task_list = [
    execute("47.93.41.197", "123456"),
    execute("47.93.41.198", "123456")
]
asyncio.run(asyncio.wait(task_list))

5.3 FastAPI框架

安裝

pip install fastapi

pip install uvicorn # (asgi 可以認為是支持異步的wsgi，內部基于uvloop)

示例：mu.py

import asyncio
import aioredis
import uvicorn
from fastapi import FastAPI
from aioredis import Redis


app = FastAPI()
REDIS_POOL = aioredis.ConnectionsPool("redis://47.193.14.198:6379", password="123", minsize=1, maxsize=10)


@app.get("/")
def index():
    """普通操作介面"""
    return {"msg": "hello world"}

@app.get("/red")
async def red():
    # 異步操作介面
    print("請求來了")
    await asyncio.sleep(3)
    # 連接池獲取一個連接
    conn = await REDIS_POOL.acquire()
    redis = Redis(conn)
    # 設定值
    await redis.hmset_dict("car", key1=1, key2=2, key3=3)
    # 讀取值
    result = await redis.hgetall("car", encoding="utf-8")
    print(result)
    # 連接歸還連接池
    REDIS_POOL.release(conn)
    return result


if __name__ == "__main__":
    uvicorn.run("mu:app", host="127.0.0.1", port=5000, log_level="info")

5.4爬蟲

pip install aiohttp

import asyncio
import aiohttp


async def fetch(session, url):
    print("發送請求", url)
    async with session.get(url, verify_ssl=False) as response:
        text = await response.text()
        print("結果：", url, len(text))
        return text


async def main():
    async with aiohttp.ClientSession() as session:
        url_list = [
            "https://python.org",
            "https://www.baidu.com",
            "https://tianbaoo.github.io"
        ]

        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        done, pending = await asyncio.wait(tasks)
        print(done)

if __name__ == "__main__":
    asyncio.run(main())

6.總結

最大的意義：通過一個執行緒利用其IO等待時間去做其他事情，

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/150701.html

標籤：Python

上一篇：用python實作在微信頭像上添加紅旗貼畫，新手必學

下一篇：Airtest常見的元素定位不到