Python自定义RabbitMQ日志处理器emit方法递归调用问题求解

我正在实现一个自定义logging handler，用于向RabbitMQ服务器发送日志。虽然已有现成工具包可实现该功能，但项目需尽量避免额外依赖。

基础实现代码 #

class RabbitMQHandler(logging.Handler):

    def __init__(self, host: str, port: int, queue: str,  level: int = 0) -> None:
        self.host = host
        self.port = port
        self.queue = queue

        super().__init__(level)


    def emit(self, record: logging.LogRecord) -> None:
        try:
            msg = self.format(record)
            with pika.BlockingConnection(pika.ConnectionParameters(
                host=self.host,
                port=self.port
            )) as connection:
                channel = connection.channel()
                channel.queue_declare(queue=self.queue)
                channel.basic_publish(
                    exchange='',
                    routing_key=self.queue,
                    body=msg
                )
            self.flush()
        except (KeyboardInterrupt, SystemExit):
            raise
        except Exception:
            self.handleError(record)

遇到的问题 #

建立连接时pika模块会自行触发日志，导致emit()函数无限递归调用。

疑问 #

是否应该在emit函数执行期间禁用日志（如何实现？），或者这种思路本身就不正确？

补充说明：将该处理器放入独立日志器是一种可行方案，但希望找到通用解决方案。

解决方案思路 #

1. 临时调整pika日志级别，屏蔽其输出 #

在emit执行期间，临时把pika模块的日志级别调高到CRITICAL以上，使其日志无法触发，执行完成后恢复原级别：

def emit(self, record: logging.LogRecord) -> None:
    # 保存pika原日志级别
    pika_logger = logging.getLogger('pika')
    original_level = pika_logger.level
    try:
        # 临时屏蔽pika所有日志输出
        pika_logger.setLevel(logging.CRITICAL + 1)
        msg = self.format(record)
        with pika.BlockingConnection(pika.ConnectionParameters(
            host=self.host,
            port=self.port
        )) as connection:
            channel = connection.channel()
            channel.queue_declare(queue=self.queue)
            channel.basic_publish(
                exchange='',
                routing_key=self.queue,
                body=msg
            )
        self.flush()
    except (KeyboardInterrupt, SystemExit):
        raise
    except Exception:
        self.handleError(record)
    finally:
        # 恢复pika原日志级别
        pika_logger.setLevel(original_level)

2. 添加过滤器，过滤pika日志 #

给自定义Handler添加过滤器，直接拒绝处理pika模块产生的日志，从根源切断递归：

class RabbitMQHandler(logging.Handler):
    def __init__(self, host: str, port: int, queue: str, level: int = 0) -> None:
        self.host = host
        self.port = port
        self.queue = queue
        super().__init__(level)
        # 绑定自定义过滤器
        self.addFilter(self._filter_pika_logs)
    
    def _filter_pika_logs(self, record: logging.LogRecord) -> bool:
        # 仅处理非pika模块的日志
        return not record.name.startswith('pika')

    def emit(self, record: logging.LogRecord) -> None:
        # 原emit逻辑保持不变
        try:
            msg = self.format(record)
            with pika.BlockingConnection(pika.ConnectionParameters(
                host=self.host,
                port=self.port
            )) as connection:
                channel = connection.channel()
                channel.queue_declare(queue=self.queue)
                channel.basic_publish(
                    exchange='',
                    routing_key=self.queue,
                    body=msg
                )
            self.flush()
        except (KeyboardInterrupt, SystemExit):
            raise
        except Exception:
            self.handleError(record)

3. 复用RabbitMQ连接（附带性能优化） #

原代码每次emit都新建连接，不仅效率低，还增加了pika触发日志的概率。可以在初始化时建立连接，后续复用，异常时重连：

class RabbitMQHandler(logging.Handler):
    def __init__(self, host: str, port: int, queue: str, level: int = 0) -> None:
        self.host = host
        self.port = port
        self.queue = queue
        self.connection = None
        self.channel = None
        super().__init__(level)
        # 初始化连接（初始化时也屏蔽pika日志）
        self._init_connection()
    
    def _init_connection(self):
        pika_logger = logging.getLogger('pika')
        original_level = pika_logger.level
        pika_logger.setLevel(logging.CRITICAL + 1)
        try:
            self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.host, port=self.port))
            self.channel = self.connection.channel()
            self.channel.queue_declare(queue=self.queue)
        finally:
            pika_logger.setLevel(original_level)
    
    def emit(self, record: logging.LogRecord) -> None:
        try:
            # 检查连接状态，失效则重连
            if not self.connection or self.connection.is_closed:
                self._init_connection()
            msg = self.format(record)
            self.channel.basic_publish(
                exchange='',
                routing_key=self.queue,
                body=msg
            )
            self.flush()
        except (KeyboardInterrupt, SystemExit):
            raise
        except Exception:
            self.handleError(record)
            # 异常时关闭连接，下次emit自动重连
            if self.connection and not self.connection.is_closed:
                self.connection.close()

思路合理性说明 #

临时禁用/过滤pika日志的思路完全可行，核心是切断pika日志触发当前Handler的链路。相比独立日志器方案，这种方式更通用，无需调整全局日志结构。优先推荐过滤器方案，因为它不会影响pika日志在其他Handler中的输出，只是当前Handler不处理；如果需要彻底屏蔽这段流程的pika日志，再用临时调整级别的方式。

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