【测试开发】python系列教程：asyncio模块

雷子

发布于 2023-09-12 14:27:09

33400

代码可运行

运行总次数：0

代码可运行

Python的asyncio模块是一个用于编写单线程并发代码的库，使用协程，多路复用IO以及其他技术。asyncio即Asynchronous I/O是python一个用来处理并发(concurrent)事件的包，是很多python异步架构的基础，多用于处理高并发网络请求方面的问题。

一、异步编程和协程

异步编程是一种编程范式，它允许程序在等待某个操作完成时继续执行其他任务。这对于IO密集型任务非常有用，例如网络请求或文件操作，这些操作通常需要等待一段时间。在Python中，我们可以使用协程（coroutines）来编写异步代码。

import asyncio

async def main():
    print('Hello')
    await asyncio.sleep(1)
    print('World')

asyncio.run(main())

结果：

其实很简单。

在asyncio中，协程不会立即执行。相反，我们需要将协程注册到事件循环（event loop），然后由事件循环来调度协程的执行。事件循环是asyncio的核心，它负责调度和执行任务。任务（Task）是对协程的封装。当我们将一个协程注册到事件循环时，它会被包装成一个任务。以下是一个创建和运行任务的例子：

import asyncio

async def hello():
    await asyncio.sleep(2)
    print('Hello, World!')

async def main():
    task = asyncio.create_task(hello())
    await task

asyncio.run(main())

结果展示

这里使用asyncio.create_task创建一个任务。

异步IO

asyncio支持异步的IO操作，包括网络请求和文件操作。

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, 'http://www.baidu.com')
        print(html)

asyncio.run(main())

看下执行结果：

异步上下文管理器和异步迭代器

asyncio还支持异步上下文管理器和异步迭代器。这允许我们在异步代码中使用with语句和for语句。异步上下文管理器是实现了异步__aenter__和__aexit__方法的对象。我们可以在async with语句中使用异步上下文管理器。异步迭代器是实现了__aiter__和__anext__方法的对象。

import asyncio

class AsyncIterable:
    def __init__(self, iterable):
        self.iterable = iterable

    def __aiter__(self):
        return self

    async def __anext__(self):
        if not self.iterable:
            raise StopAsyncIteration
        return self.iterable.pop(0)

async def main():
    async for i in AsyncIterable([100, 200, 300, 400, 500]):
        await asyncio.sleep(1)
        print(i)

asyncio.run(main())

结果：

创建协程事件对象，通过get_event_loop方法获取事件循环对象，通过run_until_complete方法直接运行event，最后结束运行

import asyncio
import time


async def async_function():  # async修饰的异步函数，在该函数中可以添加await进行暂停并切换到其他异步函数中
    now_time = time.time()
    await asyncio.sleep(1)  # 当执行await future这行代码时（future对象就是被await修饰的函数），首先future检查它自身是否已经完成，如果没有完成，挂起自身，告知当前的Task（任务）等待future完成。
    print('花费时间：{}秒'.format(time.time()-now_time))
event = async_function()  # 创建协程事件对象
loop = asyncio.get_event_loop()  # 通过get_event_loop方法获取事件循环对象
loop.run_until_complete(event)  # 通过run_until_complete方法直接运行event，该方法会一直等待直到event运行完毕
loop.close()  # 结束循环

结果

除了这些，还可以获取task的状态

import asyncio
import time

async def async_function():  # async修饰的异步函数，在该函数中可以添加await进行暂停并切换到其他异步函数中
    now_time = time.time()
    await asyncio.sleep(1)  # 当执行await future这行代码时（future对象就是被await修饰的函数），首先future检查它自身是否已经完成，如果没有完成，挂起自身，告知当前的Task（任务）等待future完成。
    print('花费时间：{}秒'.format(time.time()-now_time))
event = async_function()  # 创建协程事件对象
loop = asyncio.get_event_loop()  # 通过get_event_loop方法获取事件循环对象
task = loop.create_task(event)  # 创建任务对象
print(task)  # 任务运行中task
print(task.get_name())#任务的名称
loop.run_until_complete(task)  # 等待task运行完毕
print(task)  # 任务运行结束task状态
print(task.done())#结束
loop.close()  # 结束循环

结果，看到随机的了一个任务名称，任务的状态是running，执行后是done。

执行完毕的任务，想要在任务结果后，执行一些事情，如何获取呢？其实可以利用add_done_callback来执行回调函数。

import asyncio
import time
def task_callback(future):  # 回调函数获取任务完成后的返回值
    print("拿到结果")
    print(future.result())
async def async_function():  # async修饰的异步函数，在该函数中可以添加await进行暂停并切换到其他异步函数中
    now_time = time.time()
    await asyncio.sleep(1)  # 当执行await future这行代码时（future对象就是被await修饰的函数），首先future检查它自身是否已经完成，如果没有完成，挂起自身，告知当前的Task（任务）等待future完成。
    print('花费时间：{}秒'.format(time.time()-now_time))
event = async_function()  # 创建协程事件对象
loop = asyncio.get_event_loop()  # 通过get_event_loop方法获取事件循环对象
task = loop.create_task(event)  # 创建任务对象
task.add_done_callback(task_callback)# 为而任务添加回调函数
loop.run_until_complete(task)  # 等待task运行完毕
loop.close()  # 结束循环

如何获取到任务得返回结果呢？

用gather()收集返回值

import asyncio
import time
async def async_function(num):  # async修饰的异步函数，在该函数中可以添加await进行暂停并切换到其他异步函数中
    now_time = time.time()
    await asyncio.sleep(num)  # 当执行await future这行代码时（future对象就是被await修饰的函数），首先future检查它自身是否已经完成，如果没有完成，挂起自身，告知当前的Task（任务）等待future完成。
    print('花费时间：{}秒'.format(time.time()-now_time))
async def main():  # 异步主函数用于调度其他异步函数
    tasks = []  # tasks列表用于存放task
    for num in range(1, 4):
        tasks.append(asyncio.create_task(async_function(num)))
    response = await asyncio.gather(tasks[0], tasks[1], tasks[2])  # 将task作为参数传入gather，等异步任务都结束后返回结果列表
    print(response)
now_time = time.time()  # 程序运行时的时间戳
asyncio.run(main())  # 用asyncio.run直接运行协程参数为协程函数及其参数
print('最终执行时间：{}'.format(time.time() - now_time))

结果：