RabbitMQ Pika：SelectConnection隐藏错误细节，如何获取完整异常信息？

在使用Pika的SelectConnection时，所有错误都会被框架捕获，只能通过on_close_callback回调获取异常信息，但回调收到的始终是StreamLostError，原始异常的堆栈跟踪被隐藏，给调试带来极大不便。以下是几种解决思路，要么让异常直接抛至顶层，要么保留完整回溯信息：

问题复现示例 #

用户提供的测试代码：

import pika

class AsyncServer():
    def __init__(self):
        self._channel = None

    def run(self):
        self._connection = self._connect()
        self._connection.ioloop.start()

    def _connect(self) -> pika.SelectConnection:
        return pika.SelectConnection(
            parameters=pika.ConnectionParameters(host='localhost'),
            on_open_callback=self._on_connection_open,
            on_close_callback=self._on_connection_closed
        )

    def _on_connection_open(self, connection):
        # 触发错误
        x = 1/0

    def _on_connection_closed(self, connection, reason):
        print("Reason for closing: ",reason)
        print("Exception class: ",reason.__class__)
        self._connection.ioloop.stop()


if __name__ == '__main__':
    recv = AsyncServer()
    recv.run()

运行后仅输出：

Reason for closing:  Stream connection lost: ZeroDivisionError('division by zero')
Exception class:  <class 'pika.exceptions.StreamLostError'>

看不到x = 1/0对应的堆栈信息。

解决方法 #

1. 回调内手动捕获异常并打印完整回溯

在可能出错的回调方法中，主动捕获异常并打印完整堆栈，再手动关闭连接传递原始异常：

import pika
import traceback

class AsyncServer():
    def __init__(self):
        self._channel = None

    def run(self):
        self._connection = self._connect()
        self._connection.ioloop.start()

    def _connect(self) -> pika.SelectConnection:
        return pika.SelectConnection(
            parameters=pika.ConnectionParameters(host='localhost'),
            on_open_callback=self._on_connection_open,
            on_close_callback=self._on_connection_closed
        )

    def _on_connection_open(self, connection):
        try:
            # 触发错误
            x = 1/0
        except Exception as e:
            # 打印完整堆栈跟踪
            traceback.print_exc()
            # 主动关闭连接，传递原始异常
            connection.close(reason=e)

    def _on_connection_closed(self, connection, reason):
        print("Reason for closing: ", reason)
        print("Exception class: ", reason.__class__)
        self._connection.ioloop.stop()


if __name__ == '__main__':
    recv = AsyncServer()
    recv.run()

运行后会先输出ZeroDivisionError的完整堆栈，再显示连接关闭信息。

2. 自定义连接类重写错误处理

通过继承SelectConnection，重写内部的异常处理方法，统一捕获并打印所有回调的原始异常：

import pika
import traceback
from pika.adapters.select_connection import SelectConnection

class DebugSelectConnection(SelectConnection):
    def _on_callback_error(self, callback, *args, **kwargs):
        # 打印完整异常回溯
        traceback.print_exc()
        # 调用父类原有逻辑完成后续处理
        super()._on_callback_error(callback, *args, **kwargs)

class AsyncServer():
    def __init__(self):
        self._channel = None

    def run(self):
        self._connection = self._connect()
        self._connection.ioloop.start()

    def _connect(self) -> DebugSelectConnection:
        return DebugSelectConnection(
            parameters=pika.ConnectionParameters(host='localhost'),
            on_open_callback=self._on_connection_open,
            on_close_callback=self._on_connection_closed
        )

    def _on_connection_open(self, connection):
        # 触发错误
        x = 1/0

    def _on_connection_closed(self, connection, reason):
        print("Reason for closing: ", reason)
        print("Exception class: ", reason.__class__)
        self._connection.ioloop.stop()


if __name__ == '__main__':
    recv = AsyncServer()
    recv.run()

此方法无需修改每个回调的代码，就能全局捕获并输出所有回调的异常堆栈。

3. 启用Pika调试日志

开启Pika的DEBUG级别日志，框架会自动输出内部错误的详细信息，包括原始异常的完整回溯：

import pika
import logging

# 配置日志级别为DEBUG
logging.basicConfig(level=logging.DEBUG)

class AsyncServer():
    def __init__(self):
        self._channel = None

    def run(self):
        self._connection = self._connect()
        self._connection.ioloop.start()

    def _connect(self) -> pika.SelectConnection:
        return pika.SelectConnection(
            parameters=pika.ConnectionParameters(host='localhost'),
            on_open_callback=self._on_connection_open,
            on_close_callback=self._on_connection_closed
        )

    def _on_connection_open(self, connection):
        # 触发错误
        x = 1/0

    def _on_connection_closed(self, connection, reason):
        print("Reason for closing: ", reason)
        print("Exception class: ", reason.__class__)
        self._connection.ioloop.stop()


if __name__ == '__main__':
    recv = AsyncServer()
    recv.run()

运行后日志中会包含原始异常的完整调用栈，方便快速定位问题。

内容的提问来源于stack exchange，提问作者tango-taylor