如何在 Celery 5 中配置 circuit breaker（熔断器）？

Celery 5引入的断路器组件主要用于broker连接异常场景下的故障熔断，避免服务雪崩，具体配置方法和示例如下：

核心配置参数 #

• circuit_breaker_max_fail：触发断路器开启的连续失败次数阈值，默认值为5
• circuit_breaker_reset_timeout：断路器进入开启状态后，等待多长时间切换为半开状态尝试恢复请求，单位为秒，默认值为30
• circuit_breaker_error_classes：需要纳入熔断计数的异常类列表，默认包含所有broker连接相关的底层异常

配置示例 #

全局基础配置 #

from celery import Celery

# 初始化Celery实例
app = Celery(
    'my_tasks',
    broker='pyamqp://guest:guest@127.0.0.1:5672//',
    backend='redis://127.0.0.1:6379/0'
)

# 配置断路器参数
app.conf.broker_transport_options = {
    # 连续8次连接失败则触发熔断
    'circuit_breaker_max_fail': 8,
    # 熔断45秒后尝试恢复连接
    'circuit_breaker_reset_timeout': 45,
    # 仅指定异常类型触发熔断计数
    'circuit_breaker_error_classes': [
        'amqp.exceptions.ConnectionClosed',
        'socket.timeout',
        'kombu.exceptions.OperationalError'
    ]
}

注意事项 #

目前Celery内置的断路器仅作用于broker连接层面的异常，如果你需要对任务内部调用第三方接口、数据库等场景做熔断，需要自行实现任务级熔断逻辑，或搭配pybreaker等第三方熔断库使用。

内容的提问来源于stack exchange，提问作者Serena Xu