前言
这篇文章是《深入理解asyncio》的第三篇,主要包含回调,多线程和在asyncio中执行同步代码。
成功回调
可以给Task(Future)添加回调函数,等Task完成后就会自动调用这个(些)回调:
在这里插入图片描述
可以看到在任务完成后执行了callback函数。我这里顺便解释一个问题,不知道有没有人注意到。
为什么之前一直推荐大家用 asyncio.create_task,但是很多例子却用了 loop.create_task?
这是因为在IPython里面支持方便的使用await执行协程,但如果直接用 asyncio.create_task会报「no running event loop」: 在这里插入图片描述Eventloop是在单进程里面的单线程中的,在IPython里面await的时候会把协程注册到一个线程的Eventloop上,但是REPL环境是另外一个线程,不是一个线程,所以会提示这个错误,即便 asyncio.events._set_running_loop(loop)设置了loop,任务可以创建倒是不能await:因为task是在线程X的Eventloop上注册的,但是await时却到线程Y的Eventloop上去执行。这部分是C实现的,可以看延伸阅读链接1。
所以现在你就会看到很多 loop.create_task的代码片段,别担心,在代码项目里面都是用 asyncio.create_task的,如果你非常想要在IPython里面使用 asyncio.create_task也不是没有办法,可以这样做: 在这里插入图片描述这样就可以啦。我解释下为什么:
IPython里面能运行await是由于loop_runner函数,这个函数能运行协程(延伸阅读链接2),默认的效果大概是 asyncio.get_event_loop().run_until_complete(coro)。为了让 asyncio.create_task正常运行我定义了新的loop_runner
通过autoawait这个magic函数就可以重新设置loop_runner
上面的报错是「no running event loop」,所以通过 events._set_running_loop(loop)设置一个正在运行的loop,但是在默认的loop_runner中也无法运行,会报「Cannot run the event loop while another loop is running」,所以重置await里面那个running的loop,运行结束再设置回去。
如果你觉得有必要,可以在IPython配置文件中设置这个loop_runner到 c.InteractiveShell.loop_runner上~
调度回调
asyncio提供了3个按需回调的方法,都在Eventloop对象上,而且也支持参数:
call_soon
在下一次事件循环中被回调,回调是按其注册顺序被调用的: 在这里插入图片描述这个例子输出的比较复杂,我挨个分析:
call_soon可以用来设置任务的结果: 用 mark_done
通过2个print可以感受到 call_soon支持参数。
最重要的就是输出部分了,首先fut.done()的结果是False,因为还没到下个事件循环,sleep(0)就可以切到下次循环,这样就会调用三个 call_soon回调,最后再看fut.done()的结果就是True,而且 fut.result()可以拿到之前在 mark_done设置的值了
call_later
这次要注意3个回调的延迟时间时间要<=sleep的,要不然还没来的回调程序就结束了
call_at
安排回调在给定的时间执行,注意这个时间要基于 loop.time() 获取当前时间 在这里插入图片描述同步代码
前面的代码都是异步的,就如sleep,需要用 asyncio.sleep而不是阻塞的 time.sleep,如果有同步逻辑,怎么;利用asyncio实现并发呢?答案是用 run_in_executor。在一开始我说过开发者创建 Future 对象情况很少,主要是用 run_in_executor,就是让同步函数在一个执行器( executor)里面运行: 在这里插入图片描述可以看到用 asyncio.gather可以把同步函数逻辑转化成一个协程,且实现了并发。这里要注意细节,就是函数a是普通函数,不能写成协程,下面的定义是错误的,不能实现并发: 在这里插入图片描述
因为 a 里面没有异步代码,就不要用 asyncdef来定义。需要把这种逻辑用 loop.run_in_executor封装到协程: 在这里插入图片描述
在这里插入图片描述
大家理解了吧?
loop.run_in_executor(None,a)这里面第一个参数是要传递 concurrent.futures.Executor实例的,传递None会选择默认的executor: 在这里插入图片描述当然我们还可以用进程池,这次换个常用的文件读写例子,并且用:
在这里插入图片描述多线程
上一个小节用的 run_in_executor就如它方法的名字所示,把协程放到了一个执行器里面,可以在一个线程池,也可以在一个进程池。另外还可以使用 run_coroutine_threadsafe在其他线程执行协程(这是线程安全的): 在这里插入图片描述这里面有几个细节要注意:
协程应该从另一个线程中调用,而非事件循环运行所在线程,所以用 asyncio.new_event_loop()新建一个事件循环
在执行协程前要确保新创建的事件循环是运行着的,所以需要用 start_loop之类的方式启动循环
接着就可以用 asyncio.run_coroutine_threadsafe执行协程a了,它返回了一个Future对象
可以通过输出感受到future一开始是pending的,因为协程a里面会sleep 1秒才返回结果
用 future.result(timeout=2)就可以获得结果,设置timeout的值要大于a协程执行时间,要不然会抛出TimeoutError
一开始我们创建的新的事件循环跑在一个线程里面,由于 loop.run_forever会阻塞程序关闭,所以需要结束时杀掉线程,所以用 call_soon_threadsafe回调函数 shutdown去停止事件循环
这里再说一下 call_soon_threadsafe,看名字就知道它是线程安全版本的 call_soon,其实就是在另外一个线程里面调度回调。BTW, 其实 asyncio.run_coroutine_threadsafe底层也是用的它。
网友评论