526互联

后端技术：asyncio异步编程

发布时间 2023-12-13 10:14:03作者: 仗剑天涯横刀笑

异步的作用：

  • 异步非阻塞、asyncio详解
  • 如框架 tornado、fastapi、django >=3.x asgi、aiohttp都是在用异步，异步可以提升性能

笔记框架：

  • 协程 -- 理论
  • asyncio模块进行异步编程 -- 理论
  • 实战代码

1. 协程

协程是人为创造，不是计算机提供。

就是用一个线程让代码切换运行

协程（Coroutine），也可以被称为微线程，是一种用户态内的上下文切换技术。简而言之，其实就是通过一个线程实现代码块相互切换执行。例如：

def func1():
  print(1)
  ...
  print(2)
  
def func2():
  print(3)
  ...
  print(4)

func1()
func2()

实现协程的方法：

  • greenlet，早期模块
  • yield关键字
  • asyncio装饰器（py3.4之后）
  • async、await关键字（py3.5之后）【推荐】

1.1 greenlet 实现

pip install greenlet

from greenlet import greenlet

def func1():
    print(1)
    gr2.switch()
    print(2)
    gr2.switch()

def func2():
    print(3)
    gr1.switch()
    print(4)

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

gr1.switch()

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.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)
    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及后续版本可用。与上面的asyncio干的事一样，是为了简洁添加了这两个

import asyncio

async def func1():
    print(1)
    await asyncio.sleep(2)  # 遇到IO耗时操作，自动切换到tasks队列中的其他任务
    print(2)

async def func2():
    print(3)
    await asyncio.sleep(2)
    print(4)

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

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

2.协程意义

在一个线程中如果遇到IO等待时间，线程不会待住不动，利用空闲的时间再去干点其他事。

实例：下载三张图片（网络IO）。

  • 普通方式（同步）

    pip install requests  # 先把模块装上

    # 摘自课件
import requests

def download_image(url):
    print("开始下载:",url)
  # 发送网络请求，下载图片
    response = requests.get(url)
    print("下载完成")
    # 图片保存到本地文件
    file_name = url.rsplit('_')[-1]
    with open(file_name, mode='wb') as file_object:
        file_object.write(response.content)

if __name__ == '__main__':
    url_list = [ 'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg', 'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg'
]
    for item in url_list:
				download_image(item)

  • 协程方式（基于协程方式的异步，就是提交不会等结果，直接进行下一个）

    pip install aiohttp  # 协程实现就得安装 aiohttp 模块

    import aiohttp
import asyncio


async def fetch(session, url):
	print("发送请求:", url)
	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)


async def main():
	async with aiohttp.ClientSession() as session:
		url_list = [
'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.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(任务)

3.2 快速上手

协程函数：定义函数时候async def 函数名就可称为。

协程对象：执行 协程函数()得到的协程对象。

# func()使用了async def则称func()为协程函数
async def func()
	pass

# result调用了协程函数，则result称为协程对象
result = func()

注：执行协程函数创建协程对象的时候，函数内部代码不会执行。

那如何执行协程函数内部代码呢？

如果想要运行协程函数内部代码，必须要将协程对象交给事件循环来处理。

下面是之前的写法，但3.7之后有方便的。

import asyncio

async def func()
	print("In func()")

result = func()

# 需要借助事件循环来执行 -- 这是之前的写法
loop = asyncio.get_event_loop()
loop.run_until_complete( result )

py3.7之后的简洁写法：

import asyncio

async def func()
	print("In func()")

result = func()

asyncio.run(result) # 直接代替了loop

3.4 await

await = (async wait的意思)

await + 可等待的对象（协程对象、Future、Task对象）

示例1:

import asyncio


async def func():
    print('come here')
    response = await asyncio.sleep(2)
    print("over, ", response)


asyncio.run( func() )

示例2:

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:

import asyncio


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


async def func():
	print("执行func函数内部代码")
  
	# 一个协程函数里可以有多个 await
	response = await others()
	print("IO请求结束，结果为：", response)
  
  response = await others()
	print("IO请求结束，结果为：", response)

asyncio.run(func())

# 举的这个例子依旧是同步编程，串行。

await 就是等待对象的值得到结果之后再继续向下走。

3.4 Task对象

Tasks are used to schedule coroutines concurrently.

When a coroutine is wrapped into a Task with functions like asyncio.create_task() the coroutine is automatically scheduled to run soon。

在事件循环中添加多个任务的 （参照3.1）

Tasks用于并发调度协程，通过 asyncio.create_task(协程对象)的方式创建Task对象，这样可以让协程加入事件循环中等待被调度执行。除了使用asyncio.create_task()函数以外，还可以使用低层级的loop.create_task()或者ensure_future()函数。手动实例化Task对象不建议。

注：asyncio.create_task()函数在py3.7中被加入。在3.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对象中并立即添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
    task1 = asyncio.create_task(func())

    # 创建协程，将协程封装到一个Task对象中并立即添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
    task2 = asyncio.create_task(func())

    print("main结束")

    # 当执行某协程遇到IO操作时，会自动化切换执行其他任务。
    # 此处的await是等待相对应的协程全都执行完毕并获取结果
    ret1 = await task1
    ret2 = await task2
    print(ret1, ret2)


asyncio.run(main())

示例2：对1的更近一步使用


import asyncio


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


async def main():
	print("main开始")

# 	task_list = [
# 		asyncio.create_task( func() ),
# 		asyncio.create_task( func() ),
# 	]
	task_list = [
		asyncio.create_task(func(), name="n1"),
		asyncio.create_task(func(), name="n2")
	]

	print("main结束")

	# 当执行某协程遇到IO操作时，会自动化切换执行其他任务。
	# 此处的await是等待相对应的协程全都执行完毕并获取结果

	done, pending = await asyncio.wait(task_list, timeout=None)
	print(done)


asyncio.run(main()) # 这里就已经创建了事件循环,参照3.2

示例3：下面是了解更简写，用2最好


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对象

A Futureis a special low-level awaitable object that represents an eventual result of an asynchronous operation.

Task继承Future，Task对象内部await结果的处理基于Future对象来的。

示例1:

async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()

    # # 创建一个任务（Future对象），这个任务什么都不干。
    fut = loop.create_future()

    # 等待任务最终结果（Future对象），没有结果则会一直等下去。
    await fut

asyncio.run(main())

示例2:

import asyncio


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


async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()

    # 创建一个任务（Future对象），没绑定任何行为，则这个任务永远不知道什么时候结束。
    fut = loop.create_future()

    # 创建一个任务（Task对象），绑定了set_after函数，函数内部在2s之后，会给fut赋值。
    # 即手动设置future任务的最终结果，那么fut就可以结束了。
    await loop.create_task(set_after(fut))

    # 等待 Future对象获取 最终结果，否则一直等下去
    data = await fut
    print(data)

asyncio.run(main())

3.5.2 concurrent.futures.Future对象

这也是一个Future对象，但是与asyncio.Future没有任何关系。

这个对象干啥的？

使用线程池、进城池实现异步操作时用到的对象。

import time
from concurrent.futures import Future
from concurrent.futures.thread import ThreadPoolExecutor
from concurrent.futures.process import ProcessPoolExecutor


def func(value):
    time.sleep(1)
    print(value)

# 创建线程池
pool = ThreadPoolExecutor(max_workers=5)

# 创建进程池
# 或 pool = ProcessPoolExecutor(max_workers=5)


for i in range(10):
    fut = pool.submit(func, i)
    print(fut)

以后写代码可能会存在交叉时间。例如：crm项目80%都是基于协程异步编程 + MySQL（不支持）【线程、进程做异步编程】。

很常用

import time
import asyncio
import concurrent.futures

def func1():
    # 某个耗时操作
    time.sleep(2)
    return "SB"

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

    # 1. Run in the default loop's executor ( 默认ThreadPoolExecutor )
    # 第一步：内部会先调用 ThreadPoolExecutor 的 submit 方法去线程池中申请一个线程去执行func1函数，并返回一个concurrent.futures.Future对象
    # 第二步：调用asyncio.wrap_future将concurrent.futures.Future对象包装为asycio.Future对象。
    # 因为concurrent.futures.Future对象不支持await语法，所以需要包装为 asycio.Future对象 才能使用。
    fut = loop.run_in_executor(None, func1)  # 这一步的作用就是将concurrent.futures.Future变成asycio.Future提供await语法，就可以使用协程异步了，所以这里可以为对MySQL的操作
    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_executor(
    #         pool, func1)
    #     print('custom thread pool', result)

    # 3. Run in a custom process pool:
    # with concurrent.futures.ProcessPoolExecutor() as pool:
    #     result = await loop.run_in_executor(
    #         pool, func1)
    #     print('custom process pool', result)

asyncio.run( main() )

案例：asyncio + 不支持异步的模块，如何结合使用？

爬虫案例：

import asyncio
import requests


async def download_image(url):
    # 发送网络请求，下载图片（遇到网络下载图片的IO请求，自动化切换到其他任务）
    print("开始下载:", url)

    loop = asyncio.get_event_loop()
    # requests模块默认不支持异步操作，所以就使用线程池来配合实现了。
    future = loop.run_in_executor(None, requests.get, url)  # 把不支持异步的模块放在这里就可以异步了！
		# run_in_executor(None, requests.get, url) 这里的意思是，把requests.get为函数，url传入requests.get函数
    
    response = await future
    print('下载完成')
    # 图片保存到本地文件
    file_name = url.rsplit('_')[-1]
    with open(file_name, mode='wb') as file_object:
        file_object.write(response.content)


if __name__ == '__main__':
    url_list = [
        'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
        'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
        'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg'
    ]

    tasks = [ download_image(url)  for url in url_list]

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

3.7 异步迭代器

什么是异步迭代器

实现了 __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):
        # await asyncio.sleep(1)
        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 item in obj:
        print(item)
        
asyncio.run( func() )

3.8 异步上下文管理器

上下文管理器？

此种对象通过定义 __aenter__() 和 __aexit__() 方法来对 async with 语句中的环境进行控制。由 PEP 492 引入。

import asyncio


class AsyncContextManager:
	def __init__(self):
        self.conn = conn
        
    async def do_something(self):
        # 异步操作数据库
        return 666

    async def __aenter__(self):
        # 异步链接数据库
        self.conn = await asyncio.sleep(1)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        # 异步关闭数据库链接
		await asyncio.sleep(1)

async def func():
    async with AsyncContextManager() 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，fastapi框架内部性能提升使用的uvloop

5. 案例实战

5.1 异步操作redis

在使用python代码操作redis时，链接/操作/断开都是网络IO。(python部署在a服务器，redis部署在b服务器)

# 安装aioredis，就可以异步操作redis
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，内部在设三个键值对，即： redis = { car:{key1:1,key2:2,key3:3}}
    await redis.hmset_dict('car', key1=1, key2=2, key3=3)

    # 网络IO操作：去redis中获取值
    result = await redis.hgetall('car', encoding='utf-8')
    print(result)

    redis.close()
    # 网络IO操作：关闭redis连接
    await redis.wait_closed()

    print("结束", address)


asyncio.run( execute('redis://47.93.4.198:6379', "root!2345") )

示例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_pool(address, password=password)

    # 网络IO操作：遇到IO会自动切换任务
    await redis.hmset_dict('car', key1=1, key2=2, key3=3)

    # 网络IO操作：遇到IO会自动切换任务
    result = await redis.hgetall('car', encoding='utf-8')
    print(result)

    redis.close()
    # 网络IO操作：遇到IO会自动切换任务
    await redis.wait_closed()

    print("结束", address)


task_list = [
    execute('redis://47.93.4.197:6379', "root!2345"),
    execute('redis://47.93.4.198:6379', "root!2345")
]

asyncio.run(asyncio.wait(task_list))

5.2 异步操作MySQL

pip install aiomysql

示例1:

import asyncio
import aiomysql


async def execute():
    # 网络IO操作：连接MySQL
    conn = await aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='123', db='mysql', )

    # 网络IO操作：创建CURSOR
    cur = await conn.cursor()

    # 网络IO操作：执行SQL
    await cur.execute("SELECT Host,User 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操作：先去连接 47.93.40.197，遇到IO则自动切换任务，去连接47.93.40.198:6379
    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("SELECT Host,User 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', "root!2345"),
    execute('47.93.40.197', "root!2345")
]

asyncio.run(asyncio.wait(task_list))