一、Broker 高可用架构设计

1.1 RabbitMQ 镜像集群方案

集群搭建步骤

# 节点1初始化
rabbitmq-server -detached
rabbitmq-plugins enable rabbitmq_management# 节点2加入集群
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app# 创建镜像策略
rabbitmqctl set_policy ha-all "^celery\." '{"ha-mode":"all","ha-sync-mode":"automatic"}'

Celery 客户端配置

app.conf.broker_url = 'amqp://user:pass@node1:5672,node2:5672,node3:5672/vhost'
app.conf.broker_failover_strategy = 'shuffle'
app.conf.broker_connection_retry_on_startup = True
app.conf.broker_heartbeat = 300  # 适当延长心跳间隔

故障转移测试场景：

import socket
from kombu import Connectiondef test_failover():with Connection('amqp://node1:5672') as conn:try:conn.connection  # 强制建立连接socket.create_connection(('node1', 5672), timeout=1).close()except ConnectionError:assert conn.connection.connected  # 验证自动切换

1.2 Redis Sentinel 方案

app.conf.broker_url = 'sentinel://:mypassword@sentinel1:26379,sentinel2:26379/0'
app.conf.broker_transport_options = {'master_name': 'mymaster','sentinel_kwargs': {'password': 'sentinel_pass'},'socket_timeout': 0.5,'retry_on_timeout': True
}

二、Worker 容错机制实现

2.1 智能重试策略

@app.task(autoretry_for=(TimeoutError, IOError),retry_backoff=30,retry_backoff_max=600,retry_jitter=True,max_retries=5,acks_late=True
)
def process_payment(order_id):if db.is_connection_lost():raise self.retry(exc=ConnectionLostError())

重试参数矩阵：

参数	推荐值	作用说明
autoretry_for	(Exception,)	自动重试的异常类型
retry_backoff	30	初始退避时间（秒）
retry_backoff_max	600	最大退避时间（秒）
retry_jitter	True	添加随机抖动避免惊群效应
max_retries	3-5	最大重试次数

2.2 死信队列（DLX）配置

from kombu import Exchange, Queuedead_letter_exchange = Exchange('dlx', type='direct')
dead_letter_queue = Queue('dead_letters', exchange=dead_letter_exchange,routing_key='dead_letter')app.conf.task_queues = [Queue('orders',exchange=Exchange('orders'),routing_key='order.process',queue_arguments={'x-dead-letter-exchange': 'dlx','x-dead-letter-routing-key': 'dead_letter'}),dead_letter_queue
]@app.task(queue='dead_letters')
def handle_failed_task(task_id, exc):logger.error(f"任务 {task_id} 最终失败: {exc}")send_alert_to_ops(task_id, exc)

三、任务幂等性设计

3.1 幂等性保障方案

from celery import Task
from django.core.cache import cachescache = caches['db']class IdempotentTask(Task):def __call__(self, *args, **kwargs):task_id = self.request.idlock_key = f'task_lock:{task_id}'# 分布式锁实现if cache.add(lock_key, '1', timeout=3600):try:return self.run(*args, **kwargs)finally:cache.delete(lock_key)else:return cache.get(f'task_result:{task_id}')@app.task(base=IdempotentTask)
def process_order(order_id):result = _execute_order(order_id)cache.set(f'task_result:{order_id}', result, 86400)return result

3.2 幂等性检查清单

数据库唯一约束
版本号控制机制
请求去重令牌
状态机校验
业务层面的幂等校验

四、高可用架构验证方案

4.1 混沌工程测试

import random
from unittest.mock import patchdef test_broker_failover():with patch('kombu.transport.pyamqp.Transport.establish_connection') as mock:mock.side_effect = ConnectionErrorresult = process_order.delay(123)assert result.get(timeout=30)  # 验证任务最终成功

4.2 监控指标验证

# 重试率告警规则
alert: HighTaskRetryRate
expr: rate(celery_task_retries_total[5m]) > 0.1
for: 10m# 死信队列监控
alert: DeadLetterQueueGrowth
expr: increase(celery_dead_letters_total[1h]) > 10

五、生产环境最佳实践

5.1 容错架构检查表

Broker 集群健康检查
Worker 节点跨AZ部署
任务超时时间合理设置
结果后端独立冗余部署
定期执行故障演练

5.2 灾难恢复方案

# 紧急消息转移脚本
celery -A proj purge -Q orders  # 清空问题队列
celery -A proj control cancel_consumer orders  # 停止消费
celery -A proj control add_consumer orders -d backup_worker@node4  # 定向恢复

六、典型场景案例分析

6.1 金融交易系统

class TransactionTask(Task):acks_late = Truereject_on_worker_lost = Truepriority = 9def on_failure(self, exc, task_id, args, kwargs, einfo):rollback_transaction(args[0])super().on_failure(exc, task_id, args, kwargs, einfo)@app.task(base=TransactionTask)
def execute_transfer(source, target, amount):if Transfer.objects.filter(txid=self.request.id).exists():return  # 幂等性检查_perform_transfer(source, target, amount)

6.2 物联网数据处理

@app.task(rate_limit='100/s',autoretry_for=(DeviceOfflineError,),retry_kwargs={'max_retries': 3, 'countdown': 5},queue='iot_high'
)
def process_sensor_data(device_id, readings):if cache.get(f'device_{device_id}_status') == 'offline':raise DeviceOfflineError()_store_readings(device_id, readings)