Celery独立应用中pythonjsonlogger配置失效问题求助

这问题我之前在整合Celery和结构化日志的时候也踩过坑，核心原因是Celery会独立初始化自己的日志系统，直接在代码里写的logging.dictConfig和structlog配置会被Celery的默认日志流程覆盖，得针对Celery的特性调整配置时机和方式。下面是具体的解决步骤：

1. 用Celery的日志信号注入配置 #

Celery提供了setup_logging信号，会在它初始化日志系统的时候触发。我们需要把日志配置逻辑绑定到这个信号上，确保Celery用的是我们的配置而非默认设置：

import logging
import structlog
from structlog import stdlib, processors, threadlocal
from celery import Celery
from celery.signals import setup_logging

# 先创建Celery实例
app = Celery('your_celery_app')
app.config_from_object('django.conf:settings', namespace='CELERY')

# 绑定日志配置到Celery的初始化信号
@setup_logging.connect
def config_celery_logging(*args, **kwargs):
    # 你的logging dict配置
    logging.config.dictConfig(
        {
            "version": 1,
            "disable_existing_loggers": False,
            "formatters": {
                "json": {
                    "format": "%(name)s %(module)s %(filename)s %(funcName)s %(lineno)d %(message)s ",
                    "class": "pythonjsonlogger.jsonlogger.JsonFormatter",
                }
            },
            "handlers": {"json": {"class": "logging.StreamHandler", "formatter": "json"}},
            "loggers": { "": {"handlers": ["json"], "level": logging.DEBUG} },
        }
    )
    # structlog的配置
    structlog.configure(
        context_class=threadlocal.wrap_dict(dict),
        logger_factory=stdlib.LoggerFactory(),
        wrapper_class=stdlib.BoundLogger,
        processors=[
            stdlib.filter_by_level,
            stdlib.add_logger_name,
            stdlib.add_log_level,
            stdlib.PositionalArgumentsFormatter(),
            processors.TimeStamper(fmt="iso"),
            processors.StackInfoRenderer(),
            processors.format_exc_info,
            processors.UnicodeDecoder(),
            stdlib.render_to_log_kwargs,
        ],
    )

2. 禁用Celery自带的日志格式配置 #

检查你的Django settings或者Celery配置文件，确保没有设置CELERY_LOG_FORMAT、CELERY_TASK_LOG_FORMAT这类参数——这些配置会优先于我们的结构化日志格式，导致pythonjsonlogger失效，直接删掉或者注释掉即可。

3. 在Celery任务中正确使用日志器 #

在任务里获取日志器的时候，要使用structlog.getLogger(__name__)，而不是Celery任务实例自带的self.log（虽然配置正确时self.log也会被覆盖，但用structlog的方式更稳妥）：

@app.task
def your_sample_task():
    log = structlog.getLogger(__name__)
    log.info('Log content', input='some input')

4. 验证依赖兼容性 #

确保python-json-logger和structlog的版本是兼容的，比如避免使用过旧的structlog版本（比如<20.0）和最新的python-json-logger，版本不匹配也可能导致格式化器无法正常工作。

为什么之前的配置会失效？ #

Django启动时的日志流程是由我们的代码主导的，但Celery启动时会先独立初始化自身的日志系统，直接在代码顶部写的配置会被Celery的默认日志设置覆盖。通过setup_logging信号，我们可以在Celery准备日志系统的关键节点注入自己的配置，从而保证结构化日志格式生效。

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