<p><code>handle_sig</code>是一个<em>回调</em>,因此它直接从事件循环运行,其异常只是通过全局钩子报告给用户。如果希望在程序中的其他地方捕获在那里引发的异常,则需要使用future将异常从<code>handle_sig</code>转移到希望注意的位置</p>
<p>要在顶层捕获异常,您可能需要引入另一种方法,我们称之为<code>async_main()</code>,它等待<em>或者</em><code>self.do_io()</code>或者之前创建的未来完成:</p>
<pre><code> def __init__(self):
self.loop = asyncio.get_event_loop()
self.done_future = self.loop.create_future()
async def async_main(self):
# wait for do_io or done_future, whatever happens first
io_task = asyncio.create_task(self.do_io())
await asyncio.wait([self.done_future, io_task],
return_when=asyncio.FIRST_COMPLETED)
if self.done_future.done():
io_task.cancel()
await self.done_future # propagate the exception, if raised
else:
self.done_future.cancel()
</code></pre>
<p>要从<code>handle_sig</code>内部引发异常,只需在future对象上<a href="https://docs.python.org/3/library/asyncio-future.html#asyncio.Future.set_exception" rel="nofollow noreferrer">set the exception</a>:</p>
<pre><code> def handle_sig(self, signum):
print(f"\npid: {os.getpid()}, Received signal: {Signals(signum).name}, raising error for exit")
self.done_future.set_exception(ShutdownApp("Exiting"))
</code></pre>
<p>最后,您修改<code>run_app</code>以将<code>self.async_main()</code>传递给<code>run_until_complete</code>,然后您就可以:</p>
<pre><code>$ python3 x.py
Starting Program
io start. Press Ctrl+C now.
^C
pid: 2069230, Received signal: SIGINT, raising error for exit
ShutdownApp caught: Exiting
Finished
</code></pre>
<p>最后,请注意,可靠地捕获键盘中断是一个<a href="https://vorpus.org/blog/control-c-handling-in-python-and-trio/" rel="nofollow noreferrer">notoriously tricky undertaking</a>问题,上面的代码可能不会涵盖所有的情况</p>