6 任务路由与负载均衡

一、任务路由核心机制

1.1 静态路由配置

# celeryconfig.py

task_routes = {
    # 精确匹配任务路径
    'payment.process_order': {'queue': 'priority_payment'},
    
    # 通配符匹配任务类型
    'report.*': {'queue': 'low_priority_reports'},
    
    # 正则表达式匹配
    re.compile(r'^video\.(encode|compress)'): {'queue': 'gpu_tasks'}
}

task_default_queue = 'default_tasks'

路由优先级规则：

1. 任务装饰器直接指定的队列
2. 精确匹配的任务路径
3. 通配符匹配规则
4. 正则表达式匹配
5. 默认队列

1.2 动态路由实现

@app.task(queue=lambda task_name, args, kwargs: 
    'urgent' if kwargs.get('priority') > 8 else 'normal')
def process_data(data, priority=5):
    # 数据处理逻辑
    pass

动态路由场景：

• 根据业务参数选择队列
• 基于时间策略的路由（如节假日切换队列）
• 根据系统负载自动调整

二、权重分配与优先级控制

2.1 RabbitMQ 优先级队列实现

from kombu import Queue

app.conf.task_queues = [
    Queue('high_priority',
          exchange=Exchange('priority', type='direct'),
          routing_key='high',
          queue_arguments={'x-max-priority': 10}),
    
    Queue('normal_priority',
          queue_arguments={'x-max-priority': 5})
]

app.conf.task_default_priority = 3

优先级执行规则：

1. 高优先级队列中的任务优先执行
2. 同队列内高优先级数值的任务先执行
3. 支持0-255的优先级范围（RabbitMQ限制）

2.2 权重分配策略

# 启动不同权重的Worker
celery -A proj worker -Q high_priority -c 16  # 高权重节点
celery -A proj worker -Q normal_priority -c 8 # 普通节点
celery -A proj worker -Q batch_tasks -c 4     # 低权重节点

权重分配矩阵：

三、预取优化与性能调优

3.1 预取机制原理

app.conf.worker_prefetch_multiplier = 4  # 默认值
app.conf.worker_concurrency = 8          # 并发Worker数

# 实际预取值 = 4 * 8 = 32

预取优化公式：

最佳预取值 = (任务平均耗时(ms) / 1000) × 并发数 × 1.2

3.2 不同场景配置建议

场景1：短任务（<100ms）

worker_prefetch_multiplier = 8
worker_concurrency = 16
# 总预取：128

场景2：长任务（>10s）

worker_prefetch_multiplier = 1
worker_concurrency = 4
# 总预取：4

场景3：混合任务

# 动态调整预取策略
from celery import current_app

@after_setup_logger.connect
def setup_prefetch(sender, **kwargs):
    if 'batch' in sender.app.conf.worker_queues:
        current_app.conf.worker_prefetch_multiplier = 16

四、实战：电商订单系统案例

4.1 路由规则设计

task_routes = {
    'order.payment_callback': {
        'queue': 'critical',
        'routing_key': 'payment.urgent',
        'priority': 9
    },
    'inventory.*': {
        'queue': 'high',
        'routing_key': 'inventory.#'
    },
    'analytics.generate_report': {
        'queue': 'low',
        'exchange': 'reports'
    }
}

4.2 Worker集群配置

# 关键业务Worker（支付相关）
celery -A proj worker -Q critical -c 32 --prefetch-multiplier=2

# 常规业务Worker（库存管理）
celery -A proj worker -Q high -c 16 --prefetch-multiplier=4

# 后台任务Worker（数据分析）
celery -A proj worker -Q low -c 8 --prefetch-multiplier=8

4.3 流量高峰应对方案

# 动态路由调整
@app.task(bind=True)
def process_order(self, order_data):
    if is_peak_hours():
        self.update_state(
            queue='critical_override',
            priority=10
        )
    # 处理订单逻辑

五、高级负载均衡策略

5.1 基于资源利用率的调度

from psutil import cpu_percent

class SmartRouter:
    def route_for_task(self, task, args, kwargs):
        if cpu_percent() > 80:
            return {'queue': 'overflow'}
        return task_routes.get(task)

app.conf.task_routes = (SmartRouter(),)

5.2 跨机房流量调度

app.conf.broker_transport_options = {
    'visibility_timeout': 600,  # 10分钟
    'queue_order_strategy': 'round_robin',
    'global_keyprefix': 'bj1_'  # 北京机房标识
}

六、监控与调试技巧

6.1 关键监控指标

# 查看队列状态
celery -A proj inspect active_queues

# 检查任务分布
celery -A proj report | grep -E 'Tasks|Queues'

# 实时监控命令
watch -n 5 "celery -A proj status"

6.2 Flower 监控配置

# 启动监控服务
celery -A proj flower --port=5555

# 高级配置示例
flower --auth=user1:password1,user2:password2 \
       --persistent=True \
       --db=/var/flower/flower.db \
       --broker_api=http://rabbitmq:15672/api/

七、最佳实践

1. 路由设计原则

2. 性能调优检查表

   • 确认任务签名是否合理
   • 验证Broker消息持久化配置
   • 检查Worker心跳间隔（建议60-300秒）
   • 测试故障转移场景下的路由表现

3. 灾难恢复方案

   # 紧急流量切换
   celery control cancel_consumer queue_name  # 停止消费问题队列
   celery purge queue_name                    # 清空问题队列
   celery control add_consumer backup_queue   # 启用备用队列
   

推荐配置模板：

# 生产环境路由配置模板
task_routes = [
    ('*.critical', {'queue': 'critical', 'priority': 9}),
    ('*.high', {'queue': 'high', 'priority': 7}),
    ('*.low', {'queue': 'low', 'priority': 3})
]

worker_prefetch_multiplier = 4
worker_concurrency = 8
task_default_priority = 5

通过合理配置任务路由与负载均衡机制，可以显著提升Celery集群的处理能力。建议结合业务特点进行压力测试，持续优化路由策略和资源分配方案。