如何使用 Ctrl-C 优雅地终止 asyncio 脚本？

Question

我已经阅读了所有 post 我能找到的关于如何使用 Ctrl-C 终止 asyncio 事件循环的脚本来优雅地处理脚本，但我无法找到其中的任何一个无需像我那样打印一个或多个回溯即可工作。答案几乎无处不在，但我无法将它们中的任何一个实现到这个小脚本中：

import asyncio
import datetime
import functools
import signal


async def display_date(loop):
    end_time = loop.time() + 5.0
    while True:
        print(datetime.datetime.now())
        if (loop.time() + 1.0) >= end_time:
            break
        await asyncio.sleep(1)


def stopper(signame, loop):
    print("Got %s, stopping..." % signame)
    loop.stop()


loop = asyncio.get_event_loop()
for signame in ('SIGINT', 'SIGTERM'):
    loop.add_signal_handler(getattr(signal, signame), functools.partial(stopper, signame, loop))

loop.run_until_complete(display_date(loop))
loop.close()

我想要的是脚本在 Ctrl-C（或通过 kill 发送的 SIGTERM/SIGINT 之后不打印任何回溯）退出。此代码打印 RuntimeError: Event loop stopped before Future completed. 在我根据以前的答案尝试过的许多其他形式中，我得到了大量其他类型的异常类和错误消息，但不知道如何修复它们。上面的代码现在是最小的，但我之前所做的一些尝试绝非如此，其中 none 是正确的。

如果您能够修改脚本以使其正常终止，我们将不胜感激解释为什么您的做法是正确方式。

Answer 1

在运行时停止事件循环将永远无效。

在这里，您需要按下 Ctrl-C，以向 Python 表明您希望自己处理它而不是显示默认堆栈跟踪。这可以用经典的 try/except:

来完成

coro = display_date(loop)
try:
    loop.run_until_complete(coro)
except KeyboardInterrupt:
    print("Received exit, exiting")

并且，对于您的用例，仅此而已！对于更真实的程序，您可能需要清理一些资源。另见

Answer 2

使用 signal 个处理程序：

import asyncio
from signal import SIGINT, SIGTERM

async def main_coro():
    try:
        await awaitable()
    except asyncio.CancelledError:
        do_cleanup()

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    main_task = asyncio.ensure_future(main_coro())
    for signal in [SIGINT, SIGTERM]:
        loop.add_signal_handler(signal, main_task.cancel)
    try:
        loop.run_until_complete(main_task)
    finally:
        loop.close()

如何使用 Ctrl-C 优雅地终止 asyncio 脚本？

How to gracefully terminate an asyncio script with Ctrl-C?

python

signals

python-asyncio

sigint