本文详解如何在 Celery 应用中正确集成 structlog，解决任务内日志丢失结构化格式、被 Celery 默认前缀污染的问题，通过官方 setup_logging 信号实现主进程与 Worker 进程日志配置同步。
本文详解如何在 celery 应用中正确集成 structlog，解决任务内日志丢失结构化格式、被 celery 默认前缀污染的问题，通过官方 `setup_logging` 信号实现主进程与 worker 进程日志配置同步。
在使用 Celery 构建异步任务系统时，统一、可检索、结构化的日志（如 JSON 格式）是生产环境可观测性的基石。然而，许多开发者会遇到这样一个典型问题：应用主进程（如 Flask/Django 启动时）的日志已成功通过 structlog 配置为结构化输出，但一旦进入 Celery Task 执行上下文，日志却突然“退化”——出现 [INFO/ForkPoolWorker-2] 等 Celery 自带前缀，且 structlog 的处理器、绑定上下文（如 task_id, worker_name）全部失效，仅剩原始 logging 的朴素输出。
根本原因在于：Celery Worker 在 fork 子进程（或启动新线程/协程）执行任务时，并不会自动继承主进程的 structlog 配置；其内部默认调用 logging.basicConfig() 初始化根 logger，覆盖了你精心设计的 structlog 链路。
✅ 正确解法：利用 Celery 官方提供的生命周期信号 setup_logging —— 它在 每个 Worker 进程初始化 logging 系统后、执行任何任务前 被精确触发，是注入自定义日志配置的黄金时机。
✅ 推荐实践：通过 setup_logging 注入 structlog
以下是一个最小可行、生产就绪的配置示例（兼容 Celery ≥ 5.0，Python ≥ 3.8）：
# celery_app.py
import structlog
import logging
from celery import Celery
# 1. 定义你的 structlog 配置（推荐提取为独立函数）
def configure_structlog():
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso", utc=True),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
# 关键：使用 JSON 渲染器 → 输出结构化 JSON
structlog.processors.JSONRenderer(),
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
# 2. Celery 实例
app = Celery("myproject")
app.config_from_object("celeryconfig") # 加载 broker/result backend 等
# 3. 【核心】注册 setup_logging 信号处理器
@app.signal("setup_logging")
def setup_logging(**kwargs):
# 禁用 Celery 默认日志配置（避免冲突）
logging.getLogger("celery").handlers.clear()
logging.getLogger("celery").propagate = False
# 重新配置 structlog（此函数在每个 Worker 进程中执行）
configure_structlog()
# 可选：为 Celery 内部 logger 显式绑定 structlog
celery_logger = structlog.get_logger("celery")
celery_logger.info("structlog initialized for Celery worker")
# celeryconfig.py broker_url = "redis://localhost:6379/0" result_backend = "redis://localhost:6379/0" # 关键：禁用 Celery 自动配置日志，交由 signal 控制 worker_hijack_root_logger = False
# tasks.py
from .celery_app import app
@app.task(bind=True) # bind=True 允许访问 self.task_id
def send_notification(self, user_id: int):
logger = structlog.get_logger("tasks.notification")
logger.info("notification_start", user_id=user_id, task_id=self.request.id)
# ... 执行耗时操作（如发邮件、调用 API）
logger.info("notification_complete", status="success")
return {"status": "ok"}
? 验证效果
启动 Worker：
celery -A celery_app worker --loglevel=INFO -c 2
执行任务后，日志将不再出现 [INFO/ForkPoolWorker-2] 前缀，而是标准 JSON 行（每行一个结构化对象）：
{"event": "notification_start", "user_id": 123, "task_id": "abc123...", "level": "info", "timestamp": "2026-07-20T10:15:22.123Z"}
{"event": "notification_complete", "status": "success", "level": "info", "timestamp": "2026-07-20T10:15:25.456Z"}
⚠️ 注意事项与最佳实践
- 不要在 __init__.py 或模块顶层直接调用 structlog.configure()：这仅影响主进程，对 fork 出的 Worker 无效；
- 务必设置 worker_hijack_root_logger = False：否则 Celery 会强制接管 root logger，导致你的配置被覆盖；
- 推荐使用 bind=True + self.request.id：可自动注入 task_id 到 structlog 上下文，便于全链路追踪；
- 如需跨进程传递上下文（如 request_id）：结合 structlog.contextvars 与 Celery 的 task_prerun/task_postrun 信号增强；
- 生产环境建议搭配日志收集器（如 Filebeat / Fluentd）：直接消费 JSON 日志流，接入 ELK 或 Loki。
通过 setup_logging 信号，你无需侵入 Celery 源码、不依赖 hack 式 monkey patch，即可实现主进程与所有 Worker 进程日志行为完全一致——这才是结构化日志在分布式任务系统中落地的健壮之道。
