asyncio异步模块的21个协程编写实例

2020 年 8 月 25 日
笔记
Python

启动一个无返回值协程

通过async关键字定义一个协程

import sys
import asyncio

async def coroutine():
    print('运行协程')
    
if sys.version_info >= (3, 7, 0):
    asyncio.run(coroutine())
else:
    loop = asyncio.get_event_loop()
    loop.run_until_complete(coroutine())
    loop.close()

输出结果如下

运行协程

启动一个有返回值协程

import sys
import asyncio

async def coroutine():
    print('运行协程')
    return 'done!'
  
if sys.version_info >= (3, 7, 0):
    result = asyncio.run(coroutine())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(coroutine())
    loop.close()
print(result)

输出结果如下

运行协程
done!

使用await关键字

await关键字只能在一个协程内部使用

await等待另一个协程任务结果，它不会阻塞事件循环，但会在阻塞当前协程的上下文

import sys
import asyncio

async def coroutine():
    print('运行协程 调度asyncio.sleep协程任务')
    await asyncio.sleep(2)
    return 'done!'
  
print('启动时间:', time.time())
if sys.version_info >= (3, 7, 0):
    result = asyncio.run(coroutine())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(coroutine())
    loop.close()
print(result)
print('结束时间:', time.time())

输出结果如下

启动时间: 1598236154.90588
运行协程 调度asyncio.sleep协程任务
done!
结束时间: 1598236156.9118028

回调函数: 迅速回调

迅速回调是立即主动调用一个方法

回调方法必然是一个普通函数

回调通过事件循环loop提供的方法添加，它不会阻塞协程上下文

同一个协程内的回调函数之间存在阻塞关系，不同协程内的回调函数之间无阻塞关系

import sys
import asyncio

def callback(a, c=2):
    time.sleep(c)
    print(f"传入参数a={a}, c={c}, 时间{time.time()}")

async def main():
    print(f'添加回调, 时间{time.time()}')
    loop = asyncio.get_running_loop()
    loop.call_soon(callback, 1)
    loop.call_soon(callback, 2, 0)
    print(f'协程结束, 时间{time.time()}')

if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

输出结果如下

添加回调, 时间1598236958.119624
协程结束, 时间1598236958.1196742
传入参数a=1, c=2, 时间1598236960.1198878
传入参数a=2, c=0, 时间1598236960.119946

回调函数:延时回调

延时回调可以指定在多久后执行回调函数

延时回调的其他特性和迅速回调一致

特别注意，如果事件循环内的协程已经运行结束，尚在等待调用的回调函数不会被调用
```
import sys
import asyncio

def callback(a, c=3):
    time.sleep(c)
    print(f"传入参数a={a}, c={c}, 时间{time.time()}")

async def main():
    print(f'添加回调, 时间{time.time()}')
    loop = asyncio.get_running_loop()
    loop.call_later(1, callback, 2, 0)
    loop.call_soon(callback, 1)

    print(f'协程结束, 时间{time.time()}')

if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()
```
输出结果如下
```
添加回调, 时间1598237313.375363
协程结束, 时间1598237313.375415
传入参数a=1, c=3, 时间1598237316.376031
传入参数a=2, c=0, 时间1598237316.376389
```
解释说明，对于协程main，call_later和call_soon是不会阻塞协程上下文的，因此两个回调函数可以视为同时加入了事件循环的回调，其中回调callback(1)方法是立即调用，回调函数会sleep 3秒，而回调方法callback(2, 0)是在协程延时 1秒后调用，但回调函数之间是存在阻塞关系的，因此它会等待callback(1)先运行结束，然后判断是否满足了延时条件再执行。

回调函数:定时回调

此处的指定时间并非系统时间戳，而是指事件循环创建的时间戳

通过loop.time()获取事件循环时间戳

import sys
import asyncio

def callback(a, loop, c=3):
    time.sleep(c)
    print(f"传入参数a={a}, c={c}, 时间{loop.time()}")


async def main():
    loop = asyncio.get_running_loop()
    now = loop.time()
    print(f'事件循环时间戳{now}')
    loop.call_at(now + 4, callback, 2, loop, 1)
    loop.call_soon(callback, 1, loop)
    await asyncio.sleep(5)
    print(f'协程结束, 时间戳{loop.time()}')

if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

asyncio.Future对象

Future的实例化对象可以认为是一个协程对象，它可以使用await关键字

通过延时回调+Future可以模拟一个协程的逻辑

即延时回调 ≈ 异步等待返回结果

Future ≈ 非阻塞模型，它不会阻塞事件循环的其他协程

import sys
import asyncio

def mark_done(future: asyncio.Future, result):
    print(f'标记future结束  时间{time.time()}')
    future.set_result(result)


async def main():
    print(f'协程开始  时间{time.time()}')
    loop = asyncio.get_running_loop()
    future = asyncio.Future()
    loop.call_later(3, mark_done, future, 'done!')
    result = await future
    print(f'future结果 {result}  时间{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

输出结果如下

协程开始  时间1598238858.6768472
标记future结束  时间1598238861.678293
future结果 done!  时间1598238861.678438

Future回调

Future的回调是在一个Future对象标记结束后运行的函数

await会等待回调函数执行结束，即回调函数会在协程上下文阻塞

import sys
import asyncio

def callback(future: asyncio.Future):
    # 基于Future的回调函数一定要接受一个Future对象
    print(f'Future回调被调用 时间{time.time()}')
    time.sleep(1)


async def main():
    print(f'协程开始  时间{time.time()}')
    loop = asyncio.get_running_loop()
    future = asyncio.Future()
    future.add_done_callback(callback)
    loop.call_later(1, lambda future: future.set_result('done'), future)
    result = await future
    print(f'future结果 {result} 时间{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

输出结果如下

协程开始  时间1598241100.534262
Future回调被调用 时间1598241101.539032
future结果 done 时间1598241102.5411909

aysncio.Task对象

Task对象在调用create_task会立即执行，它类似迅速回调

asyncio.create_task是python3.7引入的高级api，3.7以下使用ensure_future方法

与回调函数的区别:

回调函数一定不是协程，而Task对象只能创建协程任务
回调函数的传入参数直接通过添加回调的方法传入，而任务对象直接传入协程形参
回调函数之间运行是阻塞的，而Task对象则是基于事件循环的标准协程

在协程上下文中可以使用await关键字等待任务结果

import sys
import asyncio

async def task_func(n):
    print(f'运行 task_func  时间{time.time()}')
    await asyncio.sleep(n)
    return 'task done!'


async def main():
    loop = asyncio.get_running_loop()
    task = loop.create_task(task_func(3))
    result = await task
    print(f'task结果 {result} 时间{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

输出结果如下

运行 task_func  时间1598241581.334356
task结果 task done! 时间1598241584.338337

Task取消任务

可以取消一个正在事件循环内运行的task对象

Task和Future一样，支持通过add_done_callback添加回调函数

import sys
import asyncio

async def task_func(n):
    print(f'运行 task_func  时间{time.time()}')
    await asyncio.sleep(n)
    return 'task done!'


async def main():
    loop = asyncio.get_running_loop()
    task = asyncio.create_task(task_func(5))
    loop.call_later(3, lambda task: task.cancel(), task)
    # Task被取消 会抛出CancelledError异常
    try:
        result = await task
        print(f'task结果 {result} 时间{time.time()}')
    except asyncio.exceptions.CancelledError:
        print(f'task被取消了 时间{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

输出结果如下

运行 task_func  时间1598242256.3299708
task被取消了 时间1598242259.335436

使用asyncio.ensure_future创建任务

asyncio.ensure_future实现效果和asyncio.create_task一致

import sys
import asyncio

async def task_func(n):
    print(f'运行 task_func  时间{time.time()}')
    await asyncio.sleep(n)
    return 'task done!'


async def main():
    print(f'协程开始  时间{time.time()}')
    task = asyncio.ensure_future(task_func(2))
    result = await task
    print(f'task结果 {result} 时间{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

输出结果如下

协程开始  时间1598249260.1346428
运行 task_func  时间1598249260.1347158
task结果 task done! 时间1598249262.137878

asyncio.wait执行多个协程

asyncio.wait接受一个协程列表/元组将其加入事件循环

返回两个列表，分别包含已完成和正在执行的Future对象

asyncio.wait 会阻塞协程上下文直至满足指定的条件（默认条件所有协程运行结束）

wait支持设置一个超时时间，但在超时发生时不会取消可等待对象，但若事件循环结束时未完成则会抛出CancelledError异常。如果要超时主动取消，可用wait_for方法

asyncio.wait 返回的结果集是按照事件循环中的任务完成顺序排列的，所以通常和原始任务顺序不同

import sys
import asyncio

async def corn_sleep(n):
    try:
        await asyncio.sleep(n)
    except asyncio.exceptions.CancelledError:
        print(f'corn({n})超时取消!  时间{time.time()}')
    print(f'corn({n}) done!  时间{time.time()}')
    return f'corn({n}) done!  时间{time.time()}'


async def main():
    print(f'协程开始  时间{time.time()}')
    tasks = [corn_sleep(i) for i in range(1, 6)]
    done, pending = await asyncio.wait(tasks, timeout=3)
    for task in done:
        pass
        # print(task.result())
    for task in pending:
        print(task.done())
    await asyncio.sleep(1)
    print(f'协程结束  时间{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

输出结果如下

协程开始  时间1598253388.614105
corn(1) done!  时间1598253389.6196852
corn(2) done!  时间1598253390.6171541
False
False
False
corn(3) done!  时间1598253391.617078
corn(4) done!  时间1598253392.617147
协程结束  时间1598253392.617195
corn(5)超时取消!  时间1598253392.6173718
corn(5) done!  时间1598253392.6173818

asyncio.gather执行多个协程

gather方法和wait都可以执行多个协程，但输入和输出有所差异

输出差异，gather保证了结果列表的顺序，它是严格遵循传入任务顺序的
输入差异，wait方法接受的是一个协程列表，而gather是通过可变长参数传入协程方法的
wait支持超时设置，gather无法设置超时时间
wait返回两个列表，列表元素是Future对象，gather只返回done列表，列表元素是Future对象的result()结果

import sys
import asyncio

async def corn_sleep(n):
    await asyncio.sleep(n)
    return f'corn({n}) done!  时间{time.time()}'


async def main():
    print(f'协程开始  时间{time.time()}')
    tasks = [corn_sleep(i) for i in range(1, 6)]
    done = await asyncio.gather(*tasks)
    for result in done:
        print(result)
    print(f'协程结束  时间{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

输出结果如下

协程开始  时间1598250547.5619462
corn(1) done!  时间1598250548.567125
corn(2) done!  时间1598250549.5660028
corn(3) done!  时间1598250550.563731
corn(4) done!  时间1598250551.566098
corn(5) done!  时间1598250552.566897
协程结束  时间1598250552.566983

asyncio.as_completed执行多个协程

as_completed方法功能同wait和gather都可以用于执行多个协程

as_completed接受的是协程列表，返回的是一个迭代器，迭代元素为Future对象

as_completed方法支持超时设置，但它会在协程上下文抛出asyncio.exceptions.TimeoutError错误

as_completed方法返回结果集是无序的

import sys
import asyncio

async def corn_sleep(n):
    await asyncio.sleep(n)
    return f'corn({n}) done!  时间{time.time()}'


async def main():
    print(f'协程开始  时间{time.time()}')
    tasks = [corn_sleep(i) for i in range(1, 6)]
    for task in asyncio.as_completed(tasks, timeout=3.5):
        try:
            result = await task
            print(result)
        except asyncio.exceptions.TimeoutError:
            print(f'协程超时了  时间{time.time()}')
            break
    print(f'协程结束  时间{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

输出结果如下

协程开始  时间1598251179.489662
corn(1) done!  时间1598251180.493866
corn(2) done!  时间1598251181.4911182
corn(3) done!  时间1598251182.493823
协程超时了  时间1598251182.991565
协程结束  时间1598251182.991592

协程锁

相比线程、进程锁，协程锁似乎应用场景不那么多，比如保存文件时?

import sys
import asyncio
from functools import partial

def callback(lock: asyncio.Lock):
    print(f'释放锁  时间{time.time()}')
    lock.release()


async def corn1(x, loop, lock):
    async with lock:
        pass
    # 加锁，利用延迟回调1秒后解锁
    await lock.acquire()
    loop.call_later(1, callback, lock)
    return f'{x} done! 时间{time.time()}'


async def main():
    loop = asyncio.get_running_loop()
    lock = asyncio.Lock()
    await lock.acquire()
    print(f'加锁 时间{time.time()}')
    loop.call_later(2, callback, lock)
    corn = partial(corn1, loop=loop, lock=lock)
    results = await asyncio.gather(corn(1), corn(2), corn(3))
    for result in results:
        print(result)


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

输出结果如下

加锁 时间1598251855.600318
释放锁  时间1598251857.6041799
释放锁  时间1598251858.606617
释放锁  时间1598251859.609798
1 done! 时间1598251857.604349
2 done! 时间1598251858.606834
3 done! 时间1598251859.6099358

Event 事件对象

Event对象和Lock对象有一定的相似性，可用于同步操作

Event对象提供一个状态标记位，它是一个布尔值，只用来判断状态

import sys
import asyncio

def task_done(event: asyncio.Event):
    event.set()


async def corn(event: asyncio.Event):
    await event.wait()
    print(f'corn done!  时间{time.time()}')


async def main():
    print(f'协程开始  时间{time.time()}')
    event = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.call_later(2, task_done, event)
    await corn(event)


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

输出结果如下

协程开始  时间1598252539.2691798
corn done!  时间1598252541.273142

Queue 协程队列

与普通队列相比，协程队列有一个task_done方法，用于标记某一个任务结束，而join方法则会阻塞，直到满足从队列取出的个数等于task_done方法调用的次数

import sys
import asyncio

async def producer(queue: asyncio.Queue):
    print('生产者 上班')
    for i in range(1, 6):
        await queue.put(f'产品({i})')
        await asyncio.sleep(2)
    await queue.put(None)
    await queue.join()
    print('生产者 打烊')


async def consumer(x, queue: asyncio.Queue):
    print(f'消费者{x}号进场')
    while True:
        pt = await queue.get()
        queue.task_done()
        if pt is None:
            await queue.put(None)
            break
        else:
            print(f'{x}号 消费了 {pt}')


async def main():
    loop = asyncio.get_running_loop()
    queue = asyncio.Queue(maxsize=3)
    loop.create_task(producer(queue))
    await asyncio.wait([consumer(i, queue) for i in range(3)])

if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

输出结果如下

生产者 上班
消费者2号进场
2号 消费了 产品(1)
消费者0号进场
消费者1号进场
2号 消费了 产品(2)
0号 消费了 产品(3)
1号 消费了 产品(4)
2号 消费了 产品(5)
生产者 下班

asyncio.subprocess 异步调用子进程

asyncio提供create_subprocess_exec和create_subprocess_shell方法

前者可以调用任意程序，后者则通过shell命令行调用其他程序

import sys
import asyncio

async def get_date():
    code = 'import datetime; print(datetime.datetime.now())'
    task1 = asyncio.create_subprocess_exec(
        sys.executable, '-c', code,
        stdout=asyncio.subprocess.PIPE
    )
    task2 = asyncio.create_subprocess_shell(
        'ls -all ~/Desktop', stdout=asyncio.subprocess.PIPE
    )
    results = await asyncio.gather(task1, task2)
    response = []
    for proc in results:
        data = await proc.stdout.read()
        line = data.decode('utf8').rstrip()
        response.append(line)
    return '\n'.join(response)

date = asyncio.run(get_date())
print(f"当前时间: {date}")

输出结果如下

当前时间: 2020-08-24 16:11:18.143943
total 16
drwxr-xr-x   3 sw  staff    96 Apr  6 09:11 $RECYCLE.BIN
drwx------@  8 sw  staff   256 Jun 18 10:52 .
drwxr-xr-x+ 63 sw  staff  2016 Aug 23 20:52 ..
-rw-r--r--@  1 sw  staff  6148 Jul 22 17:54 .DS_Store
-rw-r--r--   1 sw  staff     0 May 27  2019 .localized
drwxr-xr-x   3 sw  staff    96 Feb  3  2020 Don't Starve Together.app
drwxr-xr-x   9 sw  staff   288 May  3 18:24 HeavenMS
drwxr-xr-x   3 sw  staff    96 Sep 20  2019 Tomb Raider.app

网络通信-高级API

asyncio封装了几个高级方法来快速实现网络通信

await asyncio.open_connection() 建立TCP连接
await asyncio.open_unix_connection() 建立Unix socket连接
await start_server() 启动TCP服务
await start_unix_server() 启动Unix socket服务
StreamReader 接收网络数据的高级async/await对象
StreamWriter 发送网络数据的高级async/await对象

以搭建一个简单http服务器为例

import sys
import asyncio

async def http_handle(render: asyncio.StreamReader,
                      writer: asyncio.WriteTransport):
    """在asyncio.start_server传入该协程
    当一个tcp连接建立时就会调用该回调，并传入两个参数
    :param render: StreamReader对象
    :param writer: StreamWriter对象
    StreamReader和StreamWriter对象都是继承于Transport
    :return:
    """
    message = await render.read(1024)
    print(message.decode())
    writer.write(b'HTTP/1.1 200 OK\r\n\r\nHello Client!')
    writer.close()


async def main():
    server = await asyncio.start_server(http_handle, '127.0.0.1', 8000)
    addr = server.sockets[0].getsockname()
    print(f'服务器启动 {addr}')
    async with server:
        await server.serve_forever()


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

浏览器打开//127.0.0.1:8000 控制台输出结果如下

GET / HTTP/1.1
Host: 127.0.0.1:8000
Connection: keep-alive
Cache-Control: max-age=0
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.135 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9
Sec-Fetch-Site: none
Sec-Fetch-Mode: navigate
Sec-Fetch-User: ?1
Sec-Fetch-Dest: document
Accept-Encoding: gzip, deflate, br
Accept-Language: zh-CN,zh;q=0.9

浏览器显示结果

Hello Client!

网络通信-低级API

案例19所用的方法是基于底层Transport和Protocol封装的结果

Protocol协议类是通过重写一系列回调方法来处理网络请求

Transport实例可以简单看作是对一个socket对象封装，实现了数据收发功能

除了网络通信外，子进程（进程通信）也是通过协议回调来实现

以一个tcp服务端为例，较常用的有4个回调函数

connection_made() 连接建立时被调用
connection_lost() 连接丢失或关闭时被调用
data_received() 接收到数据时被调用
eof_received() 接收到EOF时被调用

以案例19的http服务端为例，实现相同效果

import sys
import asyncio

class HttpProtocol(asyncio.Protocol):
    def __init__(self):
        """每个tcp连接建立时都会实例化这个类"""
        self.transport: asyncio.Transport = None

    def connection_made(self, transport: asyncio.Transport):
        """连接建立时被调用"""
        self.transport = transport

    def data_received(self, data):
        """接收数据时被调用"""
        print(data.decode())
        self.transport.write(b'HTTP/1.1 200 OK\r\n\r\nHello Client!')
        self.transport.close()


loop = asyncio.get_event_loop()
server = loop.create_server(HttpProtocol, '127.0.0.1', 8000)
loop.run_until_complete(server)
loop.run_forever()

信号处理

事件循环能够支持添加信号来执行回调函数

import signal
import asyncio

def ask_exit(sig_name: str, loop: asyncio.AbstractEventLoop):
    print("捕获信号 %s: exit" % sig_name)
    loop.stop()


async def main():
    loop = asyncio.get_running_loop()
    for signame in {'SIGINT', 'SIGTERM'}:
        loop.add_signal_handler(
            getattr(signal, signame),
            partial(ask_exit, signame, loop))

    await asyncio.sleep(3600)

print("事件循环将在1小时后或者按下Ctrl+C停止")
try:
    asyncio.run(main())
except RuntimeError as e:
    if str(e) != 'Event loop stopped before Future completed.':
        raise

输出结果如下

事件循环将在1小时后或者按下Ctrl+C停止
^C捕获信号 SIGINT: exit

高频方法总结

回调和Task具有一定相似性，回调是函数，任务是协程

向事件循环非阻塞运行一个协程是创建一个任务，可以使用如下方法

方法	传入	返回	备注
asyncio.create_task	协程	Task对象	高层级api、Python3.7 +
asyncio.ensure_future	协程	Task对象	高层级api
loop.create_task	协程	Task对象	低层级api

向事件循环添加多个协程（使用await关键词才会实际添加运行），可使用如下方法

方法	传入	返回	备注
asyncio.wait	协程列表	done/pending列表元素是Future对象	支持超时设置，结果无序
asyncio.gather	变长参数	协程返回值组成的列表	返回结果列表有序
asyncio.as_completed	协程列表	协程任务迭代器	通过for循环遍历获取任务结果

在unix平台，可以使用性能更好uvloop来替代asyncio默认的事件循环
```
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
```

Tags: Python