如何在Python RQ中集成structlog日志系统以统一日志格式

很高兴能帮你搞定这个问题！要让Python RQ和你现有的structlog日志系统无缝配合，核心就是替换RQ默认的日志逻辑，让它复用你已经配置好的structlog规则，这样所有日志（不管是应用本身还是RQ的任务、worker日志）都能保持统一格式。下面是具体的实现步骤和示例：

首先，你需要把应用中已有的structlog配置整理成一个可调用的函数，这样主应用和RQ worker都能加载完全一致的规则。比如新建一个config/logging.py文件：

# config/logging.py
import structlog

def configure_structlog():
    # 这里完全复用你应用原有的structlog配置
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.JSONRenderer(),  # 替换成你实际使用的渲染器
        ],
        wrapper_class=structlog.BoundLogger,
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
    )

Python RQ允许我们通过自定义Worker类来接管它的日志行为。我们可以创建一个继承自rq.Worker的类，把它的默认logger替换成structlog的实例：

# rq_config.py
import structlog
from rq import Worker
from rq.logutils import setup_loghandlers
from config.logging import configure_structlog

# 先加载应用的structlog配置，确保和主应用一致
configure_structlog()

class StructlogWorker(Worker):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 替换RQ默认的logger为structlog的绑定日志器
        self.log = structlog.get_logger("rq.worker")
        
    def handle_exception(self, job, exc_type, exc_value, traceback):
        # 用structlog记录任务异常，同时添加Job的上下文信息（比如ID、参数）
        self.log.exception(
            "RQ任务执行失败",
            job_id=job.id,
            job_func=job.func_name,
            job_args=job.args,
            job_kwargs=job.kwargs
        )
        # 别忘了调用父类方法，保证RQ的异常处理流程正常运行
        super().handle_exception(job, exc_type, exc_value, traceback)

# 覆盖RQ默认的日志处理器初始化函数，避免重复添加标准库日志 handler
def structlog_setup_loghandlers(level=None):
    # 因为我们已经通过structlog配置好了所有日志处理，这里啥都不用做
    pass

# 替换RQ的默认setup函数
setup_loghandlers = structlog_setup_loghandlers

启动worker的时候，指定使用我们的StructlogWorker即可：

# 命令行启动方式
rq worker -w rq_config.StructlogWorker

如果是用Python代码启动worker，写法如下：

from rq import Queue
from redis import Redis
from rq_config import StructlogWorker

redis_conn = Redis(host="your-redis-host")
queue = Queue(connection=redis_conn)

# 初始化并启动自定义worker
worker = StructlogWorker([queue], connection=redis_conn)
worker.work()

在你的任务函数里，直接使用structlog记录日志就可以了，因为worker已经加载了统一的配置：

# tasks.py
import structlog

logger = structlog.get_logger("tasks")

def my_rq_task(user_id, task_data):
    # 绑定任务专属的上下文信息，方便日志追踪
    task_logger = logger.bind(user_id=user_id, task_data=task_data)
    task_logger.info("开始执行RQ任务")
    
    # 这里写你的任务逻辑...
    
    task_logger.info("RQ任务执行完成")
    return "任务结果"

• 上下文传递：如果需要把主应用的上下文（比如请求ID）传递到RQ任务中，可以用structlog.contextvars.bind_contextvars在提交任务前绑定，任务中直接读取即可。
• 日志过滤：如果要对RQ的某些系统日志做过滤，可以在structlog的processors列表中添加自定义处理器。
• 测试验证：可以写单元测试，检查RQ worker和任务的日志输出格式是否和主应用完全一致。

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