南昌哪家做网站好,wordpress标题添加新字,东营市城市和建设管理局网站,大连做网页工作室写在前面
从底层到第三方库#xff0c;全面讲解python的异步编程。这节讲述的是asyncio实现异步的上层api#xff0c;详细了解需要配合上下一节观看哦。纯干货#xff0c;无概念#xff0c;代码实例讲解。
本系列有6章左右#xff0c;点击头像或者专栏查看更多内容…写在前面
从底层到第三方库全面讲解python的异步编程。这节讲述的是asyncio实现异步的上层api详细了解需要配合上下一节观看哦。纯干货无概念代码实例讲解。
本系列有6章左右点击头像或者专栏查看更多内容陆续更新欢迎关注。
部分资料来源及参考链接 https://www.bilibili.com/video/BV1Li4y1j7RY/ https://docs.python.org/zh-cn/3.7/library/asyncio-eventloop.html
直接await协程
考虑下面一段代码
import time
import asyncioasync def one_data():print(正在暂停1号靓仔)await asyncio.sleep(3)#沉睡3秒print(正在恢复1号靓仔)async def two_data():print(正在暂停2号靓仔)await asyncio.sleep(1)#沉睡1秒print(正在恢复2号靓仔)事件循环判断两个协程的独立的,变成了同步状态
async def get_data():await one_data()#等待3秒await two_data()#等待1秒start_time time.time()#程序启动时间asyncio.run(get_data())print(程序总耗时{}.format(time.time() - start_time))在这段代码中我们仅使用了协程coroutine这个概念单单使用协程是无法完成异步编程的。因为它不能被调度虽然都写成了协程的形式但是不能同时加入一个事件循环变成了两个协程分别完成又变回同步了。
上述代码的结果就是4s多一点。表明了协程不能被直接调度局限性很高没有什么方法。所以一定要引入Task的概念实现多任务并发从而达到异步的效果。
task与协程混用
考虑以下代码
import time
import asyncioasync def one_data():print(正在暂停1号靓仔)await asyncio.sleep(3)#沉睡3秒print(正在恢复1号靓仔)async def two_data():print(正在暂停2号靓仔)await asyncio.sleep(1)#沉睡1秒print(正在恢复2号靓仔)async def one_event_loop():print(\n***我是第1个event_loop***)loop asyncio.get_event_loop()#获取当前事件循环task loop.create_task(one_data())#创建Taskawait two_data()#等待1秒await task#等待3秒start_time time.time()
asyncio.run(one_event_loop())
print(耗时{}.format(time.time() - start_time))执行结果为 这表明是可以进行混用的但是只有协程在task前面才能被识别到进行并发。如果先等待task就无法识别成功了会创造出独立的协程空间。
但是一般情况下不会混用了解一下
create_task
获取事件循环并创建是python3.5及之前的写法现在可以直接使用create_task进行创建在3.11的版本下源码是这样的
def create_task(coro, *, nameNone, contextNone):Schedule the execution of a coroutine object in a spawn task.Return a Task object.loop events.get_running_loop()if context is None:# Use legacy API if context is not neededtask loop.create_task(coro)else:task loop.create_task(coro, contextcontext)_set_task_name(task, name)return task关键的get_running_loop()官方释义是这样的 链接https://docs.python.org/zh-cn/3.7/library/asyncio-eventloop.html#asyncio.get_running_loop 返回当前 OS 线程中正在运行的事件循环。 如果没有正在运行的事件循环则会引发 RuntimeError。 此函数只能由协程或回调来调用。 那么这两个写法有区别吗
await asyncio.create_task(one_data())
await asyncio.create_task(two_data())和
task1 asyncio.create_task(one_data())
task2 asyncio.create_task(two_data())await task1
await task2答案当然是有的。
前者的写法调用create_task的时候返回task然后迅速awaittask开始运行已经有了自己的上下文。下面再次运行时上下文仍然是None又会重新创建Task。无法完成并发需求。
后者的写法第一次使用create_task后没有开始执行等待仍然存在于线程中第二次使用时就可以识别出同一上下文加入同一事件循环。开始await之后就可以达到并发效果先后顺序也没有什么关系因为都在一个事件循环中。
小总结
基于上述内容可以得到
协程不可以调度不能达到并发效果。Task与协程混用必须将协程写在Task之前可以达到并发效果但不推荐。Task调度需要保证在同一事件循环中才可完成并发。
ensure_future
或许你会问为什么不讲future呢
future类属于基类会暴露很多api ,但是在这里也可以了解一下。使用ensure_future就可以创建了。
同时这个方法也可以返回Task在函数内部有一个判断如果传入的是coroutine,task就返回task传入的是future就会返回future。
如果想深入了解 https://docs.python.org/zh-cn/3.7/library/asyncio-future.html#asyncio.ensure_future
gather
官方解释在这里https://docs.python.org/zh-cn/3.7/library/asyncio-task.html#asyncio.gather
它是这样说的 并发 运行 aws 序列中的 可等待对象。 如果 aws 中的某个可等待对象为协程它将自动作为一个任务加入日程。 aws即awaitable 如果所有可等待对象都成功完成结果将是一个由所有返回值聚合而成的列表。结果值的顺序与 aws 中可等待对象的顺序一致。 那么向这个方法传入taskcoroutine都是可以的。这个方法产生的返回值就是一个有序列表的形式。考虑一下代码
import time
import asyncioasync def one_data():print(正在暂停1号靓仔)await asyncio.sleep(3)print(正在恢复1号靓仔)return 我是1号靓仔async def two_data():print(正在暂停2号靓仔)await asyncio.sleep(1)print(正在恢复2号靓仔)return 我是2号靓仔async def get_data():print(\n***我是第1个gather***)# print(await asyncio.gather(one_data(), two_data()))使用Task写法task1 asyncio.create_task(one_data())task2 asyncio.create_task(two_data())print(await asyncio.gather(task1, task2))start_time time.time()asyncio.run(get_data())print(耗时{}.format(time.time() - start_time))运行结果 这说明顺利产生了并发进入了同一个事件循环返回值会以有序列表存储顺序为传入aws的顺序。
gather当然也提供了错误处理的方式。在常规情况下为停止模式产生错误后会立刻发送异常信号并在gather中传播代码终止。例如这样 awaitable asyncio.gather(*aws, return_exceptionsFalse) 如果 return_exceptions 为 True异常会和成功的结果一样处理并聚合至结果列表。 更改示例代码为这样
import time
import asyncioasync def one_data():print(正在暂停1号靓仔)await asyncio.sleep(3)print(正在恢复1号靓仔)return 我是1号靓仔async def two_data():print(正在暂停2号靓仔)await asyncio.sleep(1)1/0print(正在恢复2号靓仔)return 我是2号靓仔async def get_data():print(\n***我是第1个gather***)# print(await asyncio.gather(one_data(), two_data()))使用Task写法task1 asyncio.create_task(one_data())task2 asyncio.create_task(two_data())print(await asyncio.gather(task1, task2,return_exceptionsTrue))start_time time.time()asyncio.run(get_data())print(耗时{}.format(time.time() - start_time))执行结果
捕获gather异常值及任务取消
任务取消使用task.cancel()可以主动发出取消信号引发异常。异常名为CancelledError。同样可以根据设置return_exceptionsTrue来选择对异常进行停止还是忽略。
大概就像这样
task1 asyncio.create_task(one_data())
task2 asyncio.create_task(two_data())
task1.cancel()#取消print(await asyncio.gather(task1, task2,return_exceptionsTrue))执行结果 推荐写法为try-except结构
try:asyncio.run(get_data())
except asyncio.CancelledError as e:print(\n异常值{}.format(e))print(触发了CancelledError异常)3.11下的新版写法TaskGroup
gather已经是较为上层的api在3.11的版本下有了更加简洁和优雅的写法。在前面gather的使用中需要先create_task再放入 gather再await。TaskGroup可以把这几个过程再次整合。示例是这样的
async def main():async with asyncio.TaskGroup() as tg:task1 tg.create_task(some_coro(...))task2 tg.create_task(another_coro(...))print(Both tasks have completed now.)可以快速而优雅的完成并发。但是需要较高版本这里不过多介绍可以看下面的官方链接
官方链接 https://docs.python.org/zh-cn/3.11/library/asyncio-task.html#asyncio.TaskGroup
wait
官方链接 https://docs.python.org/zh-cn/3.7/library/asyncio-task.html#asyncio.wait 在3.11官方文档的解释是这样的 注意注意如果你现在使用的是3.11的版本现在已经不能直接传入协程对象那么你需要先转换为Task对象。
显然通过这个方法你可以对并发的退出条件进行颗粒化控制。不会局限于完成所有可等待任务后退出。阅读可知通过return_when来对退出条件进行控制默认为ALL_COMPLETED
代码示例
async def one_data():print(正在暂停1号靓仔)await asyncio.sleep(3)print(正在恢复1号靓仔)return 我是1号靓仔async def two_data():print(正在暂停2号靓仔)await asyncio.sleep(1)print(正在恢复2号靓仔)return 我是2号靓仔async def get_data():print(\n***我是第1个wait***)list_data [asyncio.create_task(i) for i in (one_data(), two_data())]done, pending await asyncio.wait(list_data,return_when asyncio.FIRST_COMPLETED)# done, pending await asyncio.wait(list_data,return_when asyncio.FIRST_EXCEPTION)print(\n我完成了什么{}.format(done))print(\n我没有完成什么{}.format(pending)start_time time.time()
asyncio.run(get_data())
print(\n耗时{}.format(time.time() - start_time))运行结果 可以看到可以设置首次完成任务即可退出所以耗时仅一秒Task有finishedpending状态。
设置为FIRST_EXCEPTION会在第一次出现错误的时候停止可以先设置一个错误代码。这里不再演示如果你感兴趣可以动手试试。
读取wait返回值
显然地在上述示例中返回的done,pending是set类型就像这样 直接for循环就可以读取
for i in done:#读取全部完成的任务print(\n我完成了什么{}.format(i.result()))for i in pending:#读取全部没有完成的任务print(\n我没有完成什么{}.format(i))wait_for
下面是3.11的wait_for官方解释仍然是等待aw对象完成这个方法可以从时间维度上对协程进行控制。 注意下面的版本更迭提示。
shield
shield方法用于保护任务不被取消官方文档是这样说的仍然有版本更迭注意区分 虽然可以保护取消但是在并发的时候asyncio的run方法会获取新的事件循环并指定为当前事件循环设定保护后可能会带来冲突。例如设定一个等待三秒的任务同时设有保护相当于加了一个壳还有一个等待两秒的任务取消第一个任务并使用gather进行并发。最后的结果会丢失第一个任务。这是因为外侧的任务完成后无法识别到里面的任务为同一个事件循环无法进入并发。
可以参考以下代码
import asyncioasync def one_data():print(正在暂停1号靓仔)await asyncio.sleep(1)print(正在恢复1号靓仔)return 我是1号靓仔async def two_data():print(正在暂停2号靓仔)await asyncio.sleep(2)print(正在恢复2号靓仔)return 我是2号靓仔async def mainx():task1 asyncio.shield(one_data())task2 asyncio.create_task(two_data())task1.cancel()shield asyncio.gather(task1, task2,return_exceptions True)print(await shield)asyncio.run(mainx())执行结果 最好是保证保护的任务时长比普通任务时长短可以保证保护。也可以修改最后的run方法。采用轮询的方法
loop asyncio.get_event_loop()#获取当前事件循环
loop.run_until_complete(shield())如果这里有点没理解到也没关系。后续会结合实例继续讲解。
