启动一个无返回值协程
通过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()
回调函数:延时回调
延时回调可以指定在多久后执行回调函数
延时回调的其他特性和迅速回调一致
特别注意，如果事件循环内的协程已经运行结束，尚在等待调用的回调函数不会被调用
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()
浏览器打开http://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())