celery多队列多任务的阻塞和并发问题

# celery_config.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_TASK_QUEUES = {
    'fast_tasks': {'exchange': 'fast_tasks', 'routing_key': 'fast_tasks'},
    'slow_tasks': {'exchange': 'slow_tasks', 'routing_key': 'slow_tasks'},
    'critical_tasks': {'exchange': 'critical_tasks', 'routing_key': 'critical_tasks'},
}
CELERY_TASK_ROUTES = {
    'tasks.fast_operation': {'queue': 'fast_tasks'},
    'tasks.slow_operation': {'queue': 'slow_tasks'},
    'tasks.critical_operation': {'queue': 'critical_tasks'},
}

celery -A tasks worker --queues=fast_tasks,slow_tasks,critical_tasks

@celery.task
def process_order(order_id):
    """处理订单，包含多个步骤"""
    # 步骤1：验证库存（同步阻塞）
    validate_inventory(order_id)  # 同步调用，阻塞中
    
    # 步骤2：扣减库存
    deduct_inventory(order_id)
    
    # 步骤3：生成物流单
    create_shipping(order_id)
    
    # 步骤4：发送通知
    send_notification(order_id)

@celery.task
def generate_report(report_id):
    """生成报表，需要独占数据库资源"""
    # 获取数据库锁
    with db_lock:
        # 执行复杂查询
        data = complex_database_query()
        
        # 生成报表文件
        create_report_file(data)

# 启动专门处理快速任务的高并发 Worker
celery -A tasks worker --queues=fast_tasks --concurrency=10 -n fast_worker@%h

# 启动专门处理慢速任务的低并发 Worker
celery -A tasks worker --queues=slow_tasks --concurrency=2 -n slow_worker@%h

# 启动专门处理关键任务的中等并发 Worker
celery -A tasks worker --queues=critical_tasks --concurrency=5 -n critical_worker@%h

from kombu import Queue

# 1. 定义专属队列
app.conf.task_queues = [
    Queue('default', routing_key='default'),
    Queue('sms_queue', routing_key='sms'),         # 高优先级、秒级短任务
    Queue('heavy_queue', routing_key='heavy'),     # 低优先级、耗时重任务
    Queue('high_priority', routing_key='high', queue_arguments={'x-max-priority': 10}),
    Queue('medium_priority', routing_key='medium', queue_arguments={'x-max-priority': 5}),
    Queue('low_priority', routing_key='low', queue_arguments={'x-max-priority': 1}),
]

# 2. 配置路由映射规则
app.conf.task_routes = {
    'tasks.send_sms_code': {'queue': 'sms_queue'},
    'tasks.export_large_excel': {'queue': 'heavy_queue'},
    'tasks.critical_operation': {'queue': 'high_priority'},
}

app.conf.task_default_priority = 5

# 3. 为高优先级任务装饰器
@celery.task(queue='high_priority')
def critical_task():
    """高优先级关键任务"""
    # 紧急业务逻辑
    pass

from celery import chain, group

# 错误示范 (容易引发死锁): result = task_a.delay().get()

# 正确示范 (链式调用):
@celery.task
def process_order_async(order_id):
    """异步处理订单"""
    workflow = chain(
        validate_inventory_task.s(order_id),
        deduct_inventory_task.s(order_id),
        create_shipping_task.s(order_id),
        send_notification_task.s(order_id)
    )
    return workflow.apply_async()

# 或者使用组并行执行独立任务
@celery.task
def parallel_process(order_id):
    """并行处理独立步骤"""
    independent_tasks = group([
        update_inventory.s(order_id),
        calculate_shipping.s(order_id),
        send_email_confirmation.s(order_id)
    ])
    return independent_tasks.apply_async()

# 根据任务类型配置不同的并发度
WORKER_CONFIG = {
    'image_processing': {  # CPU密集型
        'queues': ['image_queue'],
        'concurrency': 4,  # 4个进程，适合4核CPU
        'prefetch_multiplier': 1  # 减少预取，避免内存占用
    },
    'api_requests': {  # I/O密集型
        'queues': ['api_queue'],
        'concurrency': 10,  # 可以更高
        'prefetch_multiplier': 10  # 可以预取更多任务
    },
    'data_export': {  # 混合型
        'queues': ['export_queue'],
        'concurrency': 6,
        'prefetch_multiplier': 2
    }
}

# 每个 Worker 进程一次只拉取 1 个任务（针对重任务队列）
app.conf.worker_prefetch_multiplier = 1 

# 开启延迟确认，只有任务真正执行完后才从队列中删除（配合 RabbitMQ 体验更佳）
app.conf.task_acks_late = True

# 针对不同队列设置不同的预取策略
app.conf.task_queues = [
    Queue('fast_tasks', queue_arguments={'x-max-priority': 10, 'prefetch_count': 10}),
    Queue('heavy_tasks', queue_arguments={'x-max-priority': 1, 'prefetch_count': 1}),
]

# 全局超时设置
app.conf.task_time_limit = 300      # 任务执行超过 5 分钟强行杀掉
app.conf.task_soft_time_limit = 270  # 柔性超时 4.5 分钟（给任务优雅退出的缓冲时间）

# 或者针对具体任务设置超时
@celery.task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,  # 60秒后重试
    soft_time_limit=300,    # 5分钟软超时
    time_limit=360          # 6分钟硬超时
)
def long_running_task(self, data):
    """长时间运行的任务，带超时和重试"""
    try:
        # 执行可能长时间运行的操作
        result = process_large_data(data)
        return result
    except SoftTimeLimitExceeded:
        # 软超时，记录日志并重试
        self.retry(countdown=30)
    except TimeLimitExceeded:
        # 硬超时，任务被强制终止
        logger.error(f"Task {self.request.id} was terminated due to timeout")
        raise

# 启动Flower监控
celery -A tasks flower --port=5555

# 同时启动多个工作进程
celery -A tasks worker --concurrency=4 -n worker1@%h
celery -A tasks worker --concurrency=4 -n worker2@%h