asyncio应用触发KeyboardInterrupt后无法优雅关闭及AIOKafkaProducer未关闭问题

问题根因 #

• 你之前写的外层except Exception无法捕获KeyboardInterrupt异常：KeyboardInterrupt继承自BaseException而非Exception，所以普通的Exception捕获逻辑对它无效，这就是你新增外层捕获后依然有traceback的核心原因。
• 调用loop_task.cancel()后没有等待任务执行完毕：cancel()方法只会向任务发送取消信号，不会同步等待任务处理完取消逻辑，你直接去执行producer.stop()时，事件循环可能已经被提前终止，导致producer关闭逻辑没有执行完成，最终抛出未关闭的警告。
• await asyncio.gather(self.loop_task)被中断时，没有处理内部任务的取消异常，未完成的任务会被asyncio.run在终止事件循环时强制清理，不会等你自定义的stop逻辑全部走完。

修复方案 #

参考下方修正后的代码调整即可解决问题：

import asyncio
from aiokafka import AIOKafkaProducer


class Foo:
    def __init__(self):
        self.producer = None
        self.loop_task = None

    async def start(self):
        self.producer = AIOKafkaProducer(bootstrap_servers="localhost:29092")
        await self.producer.start()

        self.loop_task = asyncio.create_task(self.loop())
        try:
            await self.loop_task
        except asyncio.CancelledError:
            # 捕获任务取消异常，避免向外抛出
            pass

    async def stop(self):
        if self.loop_task and not self.loop_task.done():
            self.loop_task.cancel()
            # 等待任务处理完取消逻辑后再继续
            await self.loop_task
        if self.producer:
            await self.producer.stop()

    async def loop(self):
        while True:
            await asyncio.sleep(5)


async def main():
    foo = Foo()
    try:
        await foo.start()
    except KeyboardInterrupt:
        print(
            "Attempting graceful shutdown, press Ctrl+C again to exit...", flush=True
        )
        await foo.stop()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        # 捕获最外层可能漏出的中断信号，避免输出traceback
        pass

可选优化 #

如果担心优雅关闭过程卡住，可以在stop方法中给异步操作增加超时限制：

async def stop(self):
    if self.loop_task and not self.loop_task.done():
        self.loop_task.cancel()
        try:
            await asyncio.wait_for(self.loop_task, timeout=2)
        except (asyncio.CancelledError, asyncio.TimeoutError):
            pass
    if self.producer:
        try:
            await asyncio.wait_for(self.producer.stop(), timeout=3)
        except Exception:
            pass

内容的提问来源于stack exchange，提问作者Athena Wisdom