为何Asyncio程序中异常未终止进程反而取消任务？

我在main()函数中故意写入result = 1/0触发ZeroDivisionError，但程序并未终止，也没有任何异常捕获逻辑。反而部分待处理任务被取消，程序仍持续运行——添加该错误后服务器需要两次启动尝试（队列任务被异常取消），注释错误则首次即可启动。这个错误难道不该直接终止程序吗？

关键代码片段 #

async def main():
    # logging
    loggerTask = asyncio.create_task(initLogging(fileName=ospath.join('log','servers.log')))
    await asyncio.sleep(0)
    logger.info('servers program running')

    tasks = set()

    # socket server
    socketServer = SocketServer(None)
    tasks.add(socketServer.superviseTask)

    result = 1/0    # <----- 本应触发异常但未终止程序

    done, pending = await asyncio.wait(tasks)

完整代码 #

import os.path as ospath
import asyncio
from socket import gethostbyname, gethostname
import logging
from logging.handlers import QueueHandler, QueueListener
from queue import Queue

logger = logging.getLogger()

async def initLogging(format='%(levelname)8s - %(asctime)s - %(name)s - %(msg)s', consoleLevel=logging.INFO, fileLevel=logging.DEBUG, fileName='log.log'):
    # Change root logger level from WARNING (default) to NOTSET in order for all messages to be delegated.
    logger = logging.getLogger()
    logger.setLevel(logging.NOTSET)

    # Add file handler
    fileHandler = logging.FileHandler(filename=fileName, mode='w')
    fileHandler.setLevel(fileLevel)
    fileHandler.setFormatter(logging.Formatter(format))
    logger.addHandler(fileHandler)

    # Add queue handler
    queue = Queue()
    queueHandler = QueueHandler(queue)
    queueHandler.setLevel(logging.INFO)
    queueHandler.setFormatter(logging.Formatter(format))
    logger.addHandler(queueHandler)
    listener = QueueListener(queue, logging.StreamHandler())
    try:
        listener.start()
        logger.info('logger started')
        while True:
            await asyncio.sleep(10)
    finally:
        logger.info('logger finished')
        listener.stop

# async server base class

class AsyncServer():
    def __init__(self, app, name:str, type:str, maxRetryAttempts:int=-1, retryDelay:int=5, autoStart:bool=True) -> None:
        self.app = app
        self.name = name
        self.type = type
        self.retryCount = 0
        self.maxRetryAttempts = maxRetryAttempts # default -1 = unlimited
        self.retryDelay = retryDelay # default = 5 seconds
        if autoStart:
            self.start()

    def log(self, msg):
        return f'{self.type} {self.name} - {msg}'

    async def awaitCoro(self):
        while True:
            self.retryCount += 1
            logger.info(self.log(f'run task starting attempt {self.retryCount}/{self.maxRetryAttempts}'))
            try:
                self.runTask = asyncio.create_task(self.runCoro())
            except Exception as err:
                logger.error(self.log(f'error creating run task - {err}'))
            else:
                try:
                    await self.runTask
                    # self.retryCount = 0
                except asyncio.exceptions.CancelledError as err:
                    logger.info(self.log('run task cancelled'))
                except Exception as err:
                    logger.error(self.log(f'error - {err}'))
            # if self.runTask.done():
            #   break
            if (self.maxRetryAttempts > 0) and (self.retryCount >= self.maxRetryAttempts):
                logger.warning(self.log('max retry attempts reached'))
                break
            logger.info(self.log(f'attempting restart in {self.retryDelay} seconds'))
            await asyncio.sleep(self.retryDelay)
        logger.info(self.log('supervise task cancelled'))

    def start(self):
        logger.info(self.log('starting'))
        self.superviseTask = asyncio.create_task(self.awaitCoro())
        # self.app.taskRegistry.addItem(self.name, self.superviseTask)

    def shutdown(self):
        logger.info(self.log('shutting down'))
        self.superviseTask.cancel()
        self.runTask.cancel()
    
    def restart(self):
        logger.info(self.log('restarting'))
        self.shutdown()
        self.start()

# socket server and client

class SocketBase():
    def __init__(self, hostname:str='', ipaddr:str='', port:int=32843) -> None:
        if hostname:
            self.hostname = hostname
            self.ipaddr = gethostbyname(hostname)
        elif ipaddr:
            self.hostname = ipaddr
            self.ipaddr = ipaddr
        else:
            self.hostname = gethostname()
            self.ipaddr = gethostbyname(self.hostname)
        self.port = port

class SocketServer(SocketBase, AsyncServer):
    def __init__(self, app, hostname: str = '', ipaddr: str = '', port: int = 32843, maxRetryAttempts:int=-1, retryDelay:int=5) -> None:
        SocketBase.__init__(self, hostname, ipaddr, port)
        AsyncServer.__init__(self, app, self.hostname, 'socket server', maxRetryAttempts=maxRetryAttempts, retryDelay=retryDelay)
        self.app = app
        
    async def runCoro(self) -> None:
        try:
            self.server = await asyncio.start_server(self.handler, self.ipaddr, self.port)
            logger.info(self.log(f'listening on {self.ipaddr}:{self.port}'))
            async with self.server:
                await self.server.serve_forever()
        except Exception as err:
            logger.error(self.log(f'error starting server - {err}'))

    async def handler(self, reader:asyncio.StreamReader, writer:asyncio.StreamWriter):
        logger.info(self.log('receiving message'))

async def main():
    # logging
    loggerTask = asyncio.create_task(initLogging(fileName=ospath.join('log','servers.log')))
    await asyncio.sleep(0)
    logger.info('servers program running')

    tasks = set()
    
    # socket server
    socketServer = SocketServer(None)
    tasks.add(socketServer.superviseTask)

    result = 1/0

    done, pending = await asyncio.wait(tasks)

if __name__ == '__main__':
    asyncio.run(main())

Python 3.12.3

核心逻辑 #

asyncio.run()在处理main()协程的异常时，不会直接终止事件循环，而是先执行两步操作：

1. 取消所有由asyncio.create_task创建的未完成异步任务
2. 等待这些任务处理完取消信号后，再传播main()的异常

你的程序之所以没立刻终止，是因为两个关键任务的特殊行为：

1. SocketServer的重试机制导致任务重启

AsyncServer.awaitCoro()是一个无限循环（默认maxRetryAttempts=-1）：

• 当socketServer.superviseTask收到取消信号时，如果它正处于await self.runTask阶段，会捕获CancelledError并输出日志，但不会跳出循环
• 循环继续执行，触发下一次服务器启动尝试，这就是你看到"两次启动"的原因

2. LoggerTask未正确响应取消

initLogging()里的while True: await asyncio.sleep(10)在收到取消信号时会抛出CancelledError，进入finally块，但你写的listener.stop缺少括号，并未实际调用停止方法。即使修复这个问题，finally执行完后CancelledError会重新抛出，loggerTask最终会结束，但在此之前，superviseTask已经重启了服务器任务，让程序看起来仍在运行。

如何让异常直接终止程序 #

如果希望ZeroDivisionError触发后立刻终止程序，需要修改AsyncServer.awaitCoro()，在捕获CancelledError后跳出循环：

except asyncio.exceptions.CancelledError as err:
    logger.info(self.log('run task cancelled'))
    break  # 跳出循环，终止superviseTask

同时修复initLogging()里的listener.stop()（添加括号），确保日志监听器正确停止。

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