基于Django Channels：如何优化asyncio协程意外异常处理？

1. 用Asyncio全局异常处理器替代全局except #

这是最推荐的方式，无需修改Channels上层代码，直接给asyncio事件循环设置自定义异常处理器，所有未被捕获的协程异常都会被这个处理器接管：

import asyncio
import logging

logger = logging.getLogger(__name__)

def custom_exception_handler(loop, context):
    # 可选：先调用默认处理器保留原有基础行为
    loop.default_exception_handler(context)
    
    # 提取异常上下文信息
    exception = context.get('exception')
    task = context.get('task')
    
    if exception and task:
        logger.error(f"协程任务 {task} 抛出未捕获异常: {type(exception).__name__}: {exception}", exc_info=exception)

# 在项目启动时配置（比如Django的apps.py的ready方法，或asgi.py文件中）
loop = asyncio.get_event_loop()
loop.set_exception_handler(custom_exception_handler)

这个方案的优势：

• 覆盖所有asyncio层面的未捕获异常，包括Channels框架内部创建的任务
• 可精准提取异常上下文（任务对象、异常类型、完整堆栈），比全局except更可控
• 仅需在项目初始化时配置一次，完全无需修改Consumer子类以上的框架代码

2. 给单个任务添加完成回调 #

如果只想监控自己代码中创建的协程任务，可使用Task.add_done_callback，在任务完成时针对性检查异常：

import asyncio
import logging

logger = logging.getLogger(__name__)

def handle_task_exception(task):
    try:
        # 调用result()会触发任务中未捕获的异常
        task.result()
    except asyncio.CancelledError:
        # 任务被取消属于正常流程，无需记录
        pass
    except Exception as e:
        logger.error(f"任务 {task} 执行失败", exc_info=e)

# 创建任务时绑定回调示例
async def my_business_coroutine():
    # 你的业务逻辑代码
    pass

task = asyncio.create_task(my_business_coroutine())
task.add_done_callback(handle_task_exception)

该方式仅作用于你主动创建的任务，不会干扰框架内部任务的处理逻辑，还能针对不同任务设置差异化的异常处理规则。

3. 用装饰器包装自定义协程 #

如果你的Consumer中有自定义协程方法，可通过装饰器统一捕获异常，避免重复编写try-except：

import logging
from functools import wraps

logger = logging.getLogger(__name__)

def catch_unhandled_exceptions(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except asyncio.CancelledError:
            raise  # 必须重新抛出取消信号，保证任务能正常终止
        except Exception as e:
            logger.error(f"协程 {func.__name__} 执行异常", exc_info=e)
            # 可选：根据业务需求决定是否重新抛出异常
            # raise
    return wrapper

# 在Consumer中使用装饰器
class MyConsumer(AsyncConsumer):
    @catch_unhandled_exceptions
    async def handle_custom_event(self, event):
        # 业务逻辑代码
        pass

这个方式仅作用于你编写的代码，不会影响Channels框架的原生逻辑，同时避免了全局except的粗放式捕获。

关键注意事项 #

• 绝对不要忽略asyncio.CancelledError，这是asyncio用于任务取消的核心信号，捕获后必须重新抛出，否则会导致任务无法正常终止
• 日志中使用exc_info=e可以保留完整的异常堆栈，大幅提升排查效率
• 直接使用Django自带的logging模块即可，无需额外配置就能和项目日志系统集成

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