协程高并发 – asyncio异步

1 – 协程基础知识

• 协程函数
• 协程对象
• await关键字

2 – 执行协程函数中的代码

• 协程的顶层入口（使用run方法）
• 协程调用协程（await 一个协程对象）
• 创建task对象（使用create_task方法）

3 – 实现异步的正确方式

• 错误的调用写法
• wait方法放在顶层入口
• wait方法放在协程函数中

4 – 获取协程函数的返回值

• 使用回调函数
• 最简单的接收
• 使用wait方法
• 使用gather方法
• 使用as_completed方法

5 – 协程的高并发实例

• 执行5000个相同的协程

 # 定义协程函数 async def func(): print("这是一个协程函数") 

 # 定义协程函数 async def func(): print("这是一个协程函数") if __name__ == '__main__': # coro: 协程对象 # 此时 func函数 里面的代码 并不会像普通函数一样 被执行 coro = func() 

 import asyncio """ await 只能在协程中使用 await + 可等待的对象 比如 await 协程对象/task对象/future对象 使用create_task方法 可以把协程对象 封装成task对象 官方建议使用task对象 """ async def job(): print("正在下载") await asyncio.sleep(1) # 协程中的睡眠 print("下载完成") async def run(): print("开始执行") await job() # 此处是 await 协程对象 await asyncio.create_task(job()) # 此处是 await task对象 await asyncio.ensure_future(job()) # 此处是 await future对象 print("执行完成") 

 import asyncio # 定义协程函数 async def func(): print("这是一个协程函数") if __name__ == '__main__': # run方法: 协程的顶层入口 # 这样func函数中的代码就会执行 asyncio.run(func()) 

import asyncio async def job(param): print("job执行了, 参数是", param) # 在协程中调用另一个协程 async def run(): # await 一个协程对象 就能执行这个协程 await job("python newbee") if __name__ == '__main__': asyncio.run(run()) 

运行结果

job执行了, 参数是 python newbee 

import asyncio async def job(param): print("job执行了, 参数是", param) # 在协程中调用另一个协程 async def run(): # 把协程对象 封装成 task对象 也能执行协程 asyncio.create_task(job("task对象")) if __name__ == '__main__': asyncio.run(run()) 

运行结果

job执行了, 参数是 task对象 

import asyncio import time async def job(a): print("正在下载 参数是", a, time.time()) print("下载完成 参数是", a, time.time()) print() await asyncio.sleep(1) # 使用了3次await async def run(): coro1 = job(1) await coro1 coro2 = job(2) await coro2 coro3 = job(3) await coro3 if __name__ == '__main__': t1 = time.time() asyncio.run(run()) print("耗时:", time.time() - t1) 

运行结果

很明显是同步效果，一个任务执行消耗1秒，三个就是3秒

正在下载 参数是 1 .2335 下载完成 参数是 1 .2335 正在下载 参数是 2 . 下载完成 参数是 2 . 正在下载 参数是 3 . 下载完成 参数是 3 . 耗时: 3.00457 

import asyncio import time async def job(a): print("正在下载 参数是", a, time.time()) print("下载完成 参数是", a, time.time()) print() await asyncio.sleep(1) # 用了1次await async def run(): coros = [job(a) for a in range(3)] await asyncio.wait(coros) if __name__ == '__main__': t1 = time.time() asyncio.run(run()) print("耗时:", time.time() - t1) 

运行结果

执行的时间戳一模一样，达到了真正的异步

正在下载 参数是 1 . 下载完成 参数是 1 . 正在下载 参数是 0 . 下载完成 参数是 0 . 正在下载 参数是 2 . 下载完成 参数是 2 . 耗时: 1.00017 

import asyncio import time async def job(a): print("正在下载 参数是", a, time.time()) print("下载完成 参数是", a, time.time()) print() await asyncio.sleep(1) if __name__ == '__main__': t1 = time.time() coros = [job(a) for a in range(3)] asyncio.run(asyncio.wait(coros)) print("耗时:", time.time() - t1) 

运行结果

执行的时间戳一模一样，达到了真正的异步

正在下载 参数是 1 . 下载完成 参数是 1 . 正在下载 参数是 2 . 下载完成 参数是 2 . 正在下载 参数是 0 . 下载完成 参数是 0 . 耗时: 1.00654 

import asyncio from functools import partial # 回调函数 def callback(task): print("返回的结果是: ", task.result()) async def job(): print("job执行了") print("job结束了") return 100 async def run(): task = asyncio.create_task(job()) # 协程任务完成后, 回调函数才会执行  task.add_done_callback(partial(callback)) if __name__ == '__main__': asyncio.run(run()) 

运行结果

job执行了 job结束了 返回的结果是: 100 

import asyncio async def job(param): print("job执行了, 参数是", param) return 100 async def run(): result1 = await job("coroutine") print(result1) result2 = await asyncio.create_task(job("task")) print(result2) if __name__ == '__main__': asyncio.run(run()) 

运行结果

job执行了, 参数是 coroutine 100 job执行了, 参数是 task 100 

 import asyncio import time ''' wait方法会自动将协程对象 封装成task对象进行处理 wait方法在内部用set保存 它创建的task对象 因为set是无序的, 所以任务不是顺序执行 ''' now = lambda: time.time() async def func(idx): print(f"func执行了, 参数: { 
                                    idx}, 时间戳: { 
                                    now()}") await asyncio.sleep(1) return 100 async def run1(): tasks = [asyncio.create_task(func(idx)) for idx in range(10)] # 等待返回结果 await asyncio.wait(tasks) # 获取返回值 for task in tasks: print(task.result()) async def run2(): coros = [func(a) for a in range(10)] # wait方法会自动 把coros中的协程对象 封装成task对象 # wait方法返回一个元组, 元组中有两个集合 # done: 已经完成任务的task对象集合 # pending: 未完成任务的task对象集合 # 调用task对象的result方法可以获取 协程的返回结果 done, pending = await asyncio.wait(coros) for task in done: print(tasks.result()) if __name__ == '__main__': # 执行协程 asyncio.run(run1()) 

 import asyncio import time ''' gather方法的任务是顺序执行的 ''' now = lambda: time.time() async def func(idx): print(f"func执行了, 参数: { 
                                       idx}, 时间戳: { 
                                       now()}") return 100 async def run1(): tasks = [asyncio.create_task(func(idx)) for idx in range(10)] # 等待返回结果 await asyncio.gather(*tasks) # 获取返回值 for task in tasks: print(task.result()) async def run2(): # 把多个协程对象装到一个列表中 coros = [func(a) for a in range(10)] # gather方法返回一个列表 # 列表中的每一个元素就是协程的返回结果 results = await asyncio.gather(*coros) for result in results: print(result) if __name__ == '__main__': # 执行协程 asyncio.run(run1()) 

import asyncio from random import randint async def job(delay): print("开始下载", delay) await asyncio.sleep(delay) print("下载完成", delay) return delay async def use_as_completed(): tasks = [ asyncio.create_task(job(randint(1, 3))) for _ in range(6) ] # 使用as_completed方法的好处 # 任务先完成就立即返回, 而不是等待所有任务完成后再一起返回 for coro in asyncio.as_completed(tasks): result = await coro print("返回:", result) print("上面使用as_completed") async def use_wait(): tasks = [ asyncio.create_task(job(randint(1, 3))) for _ in range(6) ] # 使用wait方法 # 单个任务执行完了, 也要等待所有任务完成后, 再一起返回 done, pending = await asyncio.wait(tasks) for task in done: print("返回:", task.result()) print("上面使用wait") if __name__ == '__main__': asyncio.run(use_wait()) print() asyncio.run(use_as_completed()) 

运行结果

开始下载 1 开始下载 2 开始下载 1 开始下载 3 开始下载 3 开始下载 3 下载完成 1 下载完成 1 下载完成 2 下载完成 3 下载完成 3 下载完成 3 返回: 3 返回: 1 返回: 3 返回: 1 返回: 3 返回: 2 上面使用wait 开始下载 1 开始下载 3 开始下载 3 开始下载 2 开始下载 3 开始下载 1 下载完成 1 下载完成 1 返回: 1 返回: 1 下载完成 2 返回: 2 下载完成 3 下载完成 3 下载完成 3 返回: 3 返回: 3 返回: 3 上面使用as_completed 

 import asyncio import time """ 一共有5000个 parse_detail 执行 """ # 返回时间戳, 方便记录 程序执行完毕 所消耗的时间 now = lambda: time.time() # 解析详情页 async def parse_detail(id): print(f"正在解析 详情页{ 
                                          id} --> { 
                                          now()}") await asyncio.sleep(1) # 睡眠1秒 print(f"详情页{ 
                                          id} 解析完成 --> { 
                                          now()}") # 解析列表页 async def parse_page(page): print(f"正在解析 列表页{ 
                                          page} --> { 
                                          now()}") # 内层的任务列表 # 每一个页面有1000个详情页 detail_coros = [ parse_detail(f"{ 
                                          page}/{ 
                                          id}") for id in range(1, 1001) ] await asyncio.wait(detail_coros) print(f"列表页{ 
                                          page} 解析完成 --> { 
                                          now()}") async def run(): # 外层的任务列表 # 构造了5个列表页 page_coros = [ parse_page(page) for page in range(1, 6) ] await asyncio.wait(page_coros) if __name__ == '__main__': t1 = now() asyncio.run(run()) print(f"耗时: { 
                                          now() - t1}")