原文链接: 知乎文章
信号
前面的例子展示了如何用 KeyboardInterrupt 停止事件循环，即按 Ctrl-C。在 asyncio.run() 内部，引发的 KeyboardInterrupt 有效地解除对 loop.run_until_complete() 调用的阻塞，并允许随后的关闭步骤进行。
KeyboardInterrupt 对应于 SIGINT 信号。在网络服务中，更常见的进程终止信号实际上是 SIGTERM，这也是在 Unix shell 中使用 kill 命令时的默认信号。
建议：kill 命令的名称可能有误导性：它所做的一切就是向进程发送信号。如果没有加参数，kill <PID> 将发送一个 TERM 信号：你的进程可以接收到这个信号并进行优雅的关闭，或者直接忽略它。但这不是一个好主意，因为如果进程最终没有停止，另一个可能成为“杀手”的操作通常会是 kill -s KILL <PID>，它会发送 KILL 信号，强行关闭进程。
接收到 TERM（或 INT）信号是你以可控方式关闭程序的机会。
asyncio 内置了对处理进程信号的支持，但信号处理通常复杂（不仅仅是 asyncio，其他语言或库也很复杂）。我们无法涵盖所有内容，但我们可以看看需要考虑的一些更基本的问题。示例 3-33 将产生以下输出：
$ python shell_signal01.py
<Your app is running>
<Your app is running>
<Your app is running>
<Your app is running>
^C Got signal: SIGINT, shutting down.
我通过按下 Ctrl-C 停止程序，正如输出的最后一行所展示的那样。示例 3-33 有意避免使用方便的 asyncio.run() 函数，因为我想警告你在处理两个最常见的信号 SIGTERM 和 SIGINT 时的一些特定陷阱。讨论这些之后，我将展示如何使用更方便的 asyncio.run() 函数进行信号处理。
示例 3-33：使用 KeyboardInterrupt 作为 SIGINT 处理程序
# shell_signal01.py
import asyncio
async def main():
while True:
print('<Your app is running>')
await asyncio.sleep(1)
if __name__ == '__main__':
loop = asyncio.get_event_loop()
task = loop.create_task(main())
try:
loop.run_until_complete(task)
except KeyboardInterrupt: # 只会在按下 Ctrl-C 时停止循环
print('Got signal: SIGINT, shutting down.')
tasks = asyncio.all_tasks(loop=loop)
for t in tasks:
t.cancel()
group = asyncio.gather(*tasks, return_exceptions=True)
loop.run_until_complete(group)
loop.close()
示例 3-34：同时处理 SIGINT 和 SIGTERM，但只停止循环一次
# shell_signal02.py
import asyncio
from signal import SIGINT, SIGTERM
async def main():
try:
while True:
print('<Your app is running>')
await asyncio.sleep(1)
except asyncio.CancelledError:
for i in range(3):
print('<Your app is shutting down...>')
await asyncio.sleep(1)
def handler(sig):
loop.stop()
print(f'Got signal: {sig!s}, shutting down.')
loop.remove_signal_handler(SIGTERM)
loop.add_signal_handler(SIGINT, lambda: None)
if __name__ == '__main__':
loop = asyncio.get_event_loop()
for sig in (SIGTERM, SIGINT):
loop.add_signal_handler(sig, handler, sig)
loop.create_task(main())
loop.run_forever()
tasks = asyncio.all_tasks(loop=loop)
for t in tasks:
t.cancel()
group = asyncio.gather(*tasks, return_exceptions=True)
loop.run_until_complete(group)
loop.close()
执行后的输出：
CodeBlock Loading...
示例 3-35：使用 asyncio.run() 时的信号处理
CodeBlock Loading...
附：异步处理模板
CodeBlock Loading...