本文详解如何在 Celery 应用中统一启用 structlog 结构化日志，解决任务内日志丢失格式、被默认前缀污染等问题，通过 Celery 内置信号 setup_logging 实现全局日志配置自动注入。
本文详解如何在 celery 应用中统一启用 structlog 结构化日志，解决任务内日志丢失格式、被默认前缀污染等问题，通过 celery 内置信号 `setup_logging` 实现全局日志配置自动注入。
在基于 Celery 的异步任务系统中，开发者常使用 structlog 替代原生 logging 模块，以获得更清晰、可过滤、易接入 ELK 或 Datadog 等日志平台的 JSON 结构化输出。但一个常见痛点是：Celery Worker 启动后，其子进程（如 ForkPoolWorker-2）会绕过主进程的日志初始化逻辑，导致任务内日志仍走 Celery 默认的 logging 配置，丢失 structlog 的处理器、绑定上下文与 JSON 序列化能力——表现为日志前缀 [INFO/ForkPoolWorker-2] 依然存在，且内容未按预期结构化。
根本原因在于：Celery 的每个 Worker 子进程（尤其是 prefork 模式下）是独立 fork 出的 Python 进程，不会自动继承主进程的 structlog.configure() 或 logging.config.dictConfig() 调用。因此，必须在每个 Worker 进程启动时显式重置日志系统。
✅ 最佳实践：利用 Celery 官方信号 setup_logging
Celery 提供了 setup_logging 信号，专为该场景设计——它在每个 Worker 进程初始化日志系统之后、执行任何任务之前触发，允许你安全地覆盖或增强默认日志配置：
# celery_app.py
import structlog
import logging
from celery import Celery
app = Celery("myapp")
app.config_from_object("celeryconfig") # 如 broker_url, task_serializer 等
# ✅ 全局初始化 structlog（主进程）
def configure_structlog():
structlog.configure(
processors=[
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(),
structlog.processors.JSONRenderer(), # 关键：输出 JSON
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
# ✅ 在每个 Worker 进程启动时触发
@app.task(bind=True)
def dummy_task(self):
logger = structlog.get_logger()
logger.info("This task uses structlog", task_id=self.request.id, user_id=123)
# ? 注册 setup_logging 信号处理器
from celery.signals import setup_logging
@setup_logging.connect
def setup_logging_handler(sender=None, **kwargs):
# 此处会被每个 Worker 进程调用一次
configure_structlog()
# 可选：禁用 Celery 默认日志处理器，避免重复输出
root_logger = logging.getLogger()
for handler in root_logger.handlers[:]:
root_logger.removeHandler(handler)
# 将 structlog 的标准输出处理器添加到 root logger
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s")) # 让 JSON 原样输出
root_logger.addHandler(handler)
root_logger.setLevel(logging.INFO)
同时，确保你的 celeryconfig.py 中已正确配置 Broker 和基础选项：
# 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
启动 Worker 时无需额外参数，Celery 会自动触发信号：
celery -A celery_app worker --loglevel=info
? 注意事项与进阶建议：
- 避免在 @task 内部初始化：不要在每个任务函数开头调用 structlog.configure()，这会导致重复配置、性能损耗及线程不安全风险；
- 进程隔离性：setup_logging 信号在每个 forked Worker 进程中独立执行，天然适配多进程并发模型；
- 上下文绑定增强：可在 setup_logging 处理器中预绑定通用字段（如服务名、环境），或结合 Celery 的 task_prerun 信号动态注入任务元数据（如 task_id, args, kwargs）；
- 兼容 Flask/Django 集成：若 Celery 与 Web 框架共用，建议将 configure_structlog() 提取为独立模块，在 Web 启动和 setup_logging 中复用，保证全栈日志格式一致；
- 生产环境建议：搭配 RotatingFileHandler 或 SysLogHandler 替代 StreamHandler，并设置 backupCount 和 maxBytes 防止日志文件无限增长。
至此，所有 Celery 任务日志（包括 dummy_task）将输出纯净 JSON，无冗余前缀，且自动包含时间戳、日志等级、上下文字段等结构化信息，真正实现可观测性友好的一致日志体验。
