本文详解如何在 Celery 应用中统一启用 structlog 结构化日志，解决任务内日志被默认格式覆盖、上下文丢失、JSON 字段缺失等典型问题，通过 Celery 内置信号 setup_logging 实现全局日志初始化。
本文详解如何在 celery 应用中统一启用 structlog 结构化日志，解决任务内日志被默认格式覆盖、上下文丢失、json 字段缺失等典型问题，通过 celery 内置信号 `setup_logging` 实现全局日志初始化。
在基于 Celery 的分布式任务系统中，日志一致性是可观测性的基石。许多团队已成功在主应用（如 Flask/Django）中配置 structlog 实现结构化 JSON 日志输出，但一旦进入 Celery 任务上下文，日志却突然“退化”为带 [INFO/ForkPoolWorker-2] 前缀的原始格式，且关键字段（如 task_id、task_name、correlation_id）缺失——这并非 structlog 失效，而是 Celery 在 fork 子进程时未自动继承主进程的日志配置，导致每个 Worker 进程使用了默认的 logging 配置。
幸运的是，Celery 提供了开箱即用的解决方案：setup_logging 信号。该信号在每个 Worker 子进程启动日志系统后、执行任何任务前精准触发，是注入自定义日志配置（包括 structlog 初始化）的理想时机。
✅ 正确做法：利用 setup_logging 信号统一初始化
以下是一个生产就绪的完整示例（兼容 Celery ≥ 5.0，Python ≥ 3.8）：
# celery_app.py
import structlog
import logging
from celery import Celery
app = Celery("myproject")
app.config_from_object("celeryconfig") # 加载 broker/redis 等配置
# 1️⃣ 定义结构化日志配置（主进程 & Worker 进程共用）
def configure_structlog():
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars, # 支持 async context
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
# 关键：为 Celery 任务自动注入 task 上下文
structlog.processors.CallsiteParameterAdder(
[
structlog.processors.CallsiteParameter.FILENAME,
structlog.processors.CallsiteParameter.FUNC_NAME,
structlog.processors.CallsiteParameter.LINENO,
]
),
structlog.processors.JSONRenderer(), # 输出 JSON
],
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
# 2️⃣ 注册 setup_logging 信号处理器
@app.on_after_configure.connect
def setup_logging_signal(sender, **kwargs):
"""
注意：Celery 5.0+ 推荐使用 on_after_configure（更早触发）
若使用 Celery <pre class="brush:php;toolbar:false;"># celeryconfig.py
broker_url = "redis://localhost:6379/0"
result_backend = "redis://localhost:6379/0"
task_serializer = "json"
result_serializer = "json"
accept_content = ["json"]
timezone = "Asia/Shanghai"
enable_utc = True# tasks.py
from .celery_app import app
@app.task(bind=True, track_started=True)
def send_notification(self, user_id: int, message: str):
logger = structlog.get_logger()
# 自动携带 task_id、task_name、retries 等上下文
logger.info(
"notification_sent",
user_id=user_id,
message_preview=message[:50],
task_retries=self.request.retries,
)
# ... 实际业务逻辑
return {"status": "success"}⚠️ 关键注意事项
- 不要在任务函数内调用 structlog.configure()：每次调用会覆盖全局配置，引发竞态与性能损耗；
- 避免 @app.task 装饰器中 logger = structlog.get_logger() 的静态初始化：应始终在任务执行时动态获取 logger，以捕获当前 task 请求上下文；
- on_after_configure vs setup_logging：Celery 5.0+ 强烈推荐前者，它在配置加载完成后、日志系统初始化前触发，确保你的 dictConfig 或 structlog.configure() 在 Celery 默认 handler 安装前生效；
- 多进程安全：structlog 本身线程/进程安全，但若你额外添加了 FileHandler，请确保文件路径支持并发写入（推荐 RotatingFileHandler 或日志收集服务如 Loki/ELK）；
- 调试技巧：启动 worker 时添加 --loglevel=DEBUG 并观察 "Using custom logging configuration" 日志，确认信号已触发。
✅ 验证效果
启动 worker：
celery -A celery_app worker --loglevel=INFO -c 2
触发任务后，日志将输出标准 JSON（无 [INFO/ForkPoolWorker-2] 前缀）：
{
"event": "notification_sent",
"user_id": 1001,
"message_preview": "Welcome to our platform!",
"task_retries": 0,
"task_id": "a1b2c3d4-e5f6-7890-g1h2-i3j4k5l6m7n8",
"task_name": "tasks.send_notification",
"timestamp": "2026-07-20T09:24:15.123Z",
"level": "info",
"logger": "tasks"
}
至此，Celery 任务日志与主应用日志完全对齐，具备可过滤、可聚合、可追踪的结构化能力，为现代云原生可观测性打下坚实基础。
