Python 异步任务队列：Celery + Redis/RabbitMQ 实战

from celery import Celery

# 创建 Celery 应用
app = Celery('tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',
    include=['tasks.email', 'tasks.image', 'tasks.report']
)

# 从配置文件加载配置
app.config_from_object('celeryconfig')

# 定义普通任务
@app.task
def process_image(image_path: str):
    """处理图片任务"""
    from PIL import Image
    img = Image.open(image_path)
    img.thumbnail((800, 600))
    img.save(image_path + '.thumb.jpg')
    return {"thumbnail": image_path + '.thumb.jpg'}

# 绑定任务：self 可访问任务请求上下文
@app.task(bind=True, max_retries=3, default_retry_delay=60)
def send_email(self, to: str, subject: str, body: str):
    """发送邮件任务，支持重试与指数退避"""
    try:
        smtp.send(to, subject, body)
        return {"status": "sent", "to": to}
    except Exception as exc:
        # 指数退避重试：1, 2, 4, 8... 分钟
        countdown = min(2 ** self.request.retries * 60, 3600)
        raise self.retry(exc=exc, countdown=countdown)

from celery import Task
import logging

logger = logging.getLogger(__name__)

class LoggingTask(Task):
    """自定义基类：自动记录任务开始、结束与异常"""
    abstract = True

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        logger.error(f"Task {task_id} failed: {exc}")
        super().on_failure(exc, task_id, args, kwargs, einfo)

    def on_success(self, retval, task_id, args, kwargs):
        logger.info(f"Task {task_id} succeeded in {self.request.timelimit}")
        super().on_success(retval, task_id, args, kwargs)

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        logger.warning(f"Task {task_id} retrying: {exc}")
        super().on_retry(exc, task_id, args, kwargs, einfo)

@app.task(base=LoggingTask, name='tasks.generate_report')
def generate_report(user_id: int, report_type: str):
    """带日志追踪的报告生成任务"""
    logger.info(f"Generating {report_type} for user {user_id}")
    # 模拟耗时操作
    time.sleep(5)
    return {"user_id": user_id, "report_type": report_type, "done": True}

from celery import chain, chord, group, signature

# 签名：任务的延迟引用
s = send_email.s('user@example.com', 'Hello', 'Welcome!')

# Chain：顺序执行，前一个任务的结果作为下一个任务的参数
workflow = chain(
    download_image.s(url),
    process_image.s(),
    upload_to_cdn.s()
)
result = workflow.apply_async()

# Group：并行执行一组任务
g = group([
    process_image.s('img1.jpg'),
    process_image.s('img2.jpg'),
    process_image.s('img3.jpg'),
])
result = g.apply_async()  # 返回 GroupResult，可查询所有子任务状态

# Chord：先并行执行 group，等全部完成后将结果汇总传给回调
chord(
    (process_image.s(f'img{i}.jpg') for i in range(10)),
    merge_thumbnails.s()  # 回调接收所有子任务结果列表
).apply_async()

# Map / Starmap：类似 Python 内置函数
# map：每个元素作为第一个参数传入同一任务
result = process_image.map(['img1.jpg', 'img2.jpg', 'img3.jpg'])
# starmap：每个元组解包后传入
result = send_email.starmap([
    ('user1@example.com', 'Hi', 'Msg1'),
    ('user2@example.com', 'Hi', 'Msg2'),
])

from celery import Celery
from celery.schedules import crontab, timedelta

app = Celery('tasks')

# 通过 on_after_configure 信号注册定时任务
@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # 每 60 秒执行清理
    sender.add_periodic_task(60.0, cleanup.s(), name='cleanup every minute')
    # 每早 2 点执行备份
    sender.add_periodic_task(crontab(hour=2, minute=0), backup.s())
    # 每个工作日 9:30 执行日报
    sender.add_periodic_task(
        crontab(hour=9, minute=30, day_of_week='mon,tue,wed,thu,fri'),
        daily_report.s()
    )
    # 每 15 分钟执行一次，带抖动避免雪崩
    sender.add_periodic_task(
        crontab(minute='*/15'),
        heartbeat_check.s(),
        options={'countdown': random.randint(0, 30)}
    )

# 或通过 beat_schedule 配置
app.conf.beat_schedule = {
    'cleanup-every-minute': {
        'task': 'tasks.cleanup',
        'schedule': 60.0,
        'args': (),
        'options': {'queue': 'maintenance'},
    },
    'daily-backup': {
        'task': 'tasks.backup',
        'schedule': crontab(hour=2, minute=0),
    },
}

from datetime import datetime, timedelta

# delay：最简调用，位置参数传值
result = send_email.delay('user@example.com', 'Hello', 'Welcome!')
print(result.id)  # 获取任务 UUID

# 查询结果
result = send_email.AsyncResult(result.id)
if result.ready():
    print(result.get(timeout=10))  # 阻塞获取结果，超时 10 秒

# 非阻塞轮询
while not result.ready():
    time.sleep(1)
print(result.result)  # 直接访问结果属性

# apply_async：全功能调用，支持所有调度参数
send_email.apply_async(
    args=['user@example.com', 'Hello', 'Welcome!'],
    kwargs={'priority': 'high'},
    countdown=300,           # 延迟 5 分钟执行
    eta=datetime(2024, 1, 1, 10, 0),  # 指定时间点
    expires=datetime(2024, 1, 1, 12, 0),  # 过期不执行
    queue='email',           # 指定队列
    routing_key='email.high',
    task_id='custom-task-id',  # 自定义任务 ID
    retry=True,
    retry_policy={
        'max_retries': 5,
        'interval_start': 30,
        'interval_step': 30,
        'interval_max': 300,
    }
)

from celery import chain, group

# link：任务成功后的回调
result = send_email.apply_async(
    args=['user@example.com', 'Welcome', 'Thanks!'],
    link=send_notification.s('push')  # send_email 的结果传给 send_notification
)

# link_error：任务失败后的错误回调
result = process_image.apply_async(
    args=['photo.jpg'],
    link_error=notify_admin.s('image-processing-failed')
)

# 同时设置成功和失败回调
result = process_image.apply_async(
    args=['photo.jpg'],
    link=send_notification.s('success'),
    link_error=send_notification.s('failed')
)

from celery import chain

# 场景：用户注册后，发送欢迎邮件 → 初始化默认配置 → 推送通知
def register_user_workflow(user_id: str, email: str):
    workflow = chain(
        create_user.s(user_id, email),
        send_email.s('welcome@example.com', 'Welcome!', email),
        init_user_settings.s(),
        push_notification.s('welcome_push')
    )
    result = workflow.apply_async()
    return result.id

# 等效写法：管道操作符
workflow = create_user.s(user_id, email) | send_email.s() | init_user_settings.s() | push_notification.s()
result = workflow.apply_async()

# chain 中某步失败时的处理
def safe_register_workflow(user_id: str, email: str):
    workflow = chain(
        create_user.si(user_id, email),     # si 表示不接收上一步结果
        send_email.s('welcome@example.com', 'Welcome!', email),
    )
    result = workflow.apply_async(
        link_error=compensate_user_creation.s(user_id)  # 回滚操作
    )
    return result

from celery import group, chord, chain

# Group：并行处理多张图片，汇总结果
def batch_process_images(image_urls: list):
    job = group(
        download_and_process.s(url) for url in image_urls
    )
    result = job.apply_async()
    # 等待所有任务完成
    results = result.get(timeout=120)  # 返回列表，顺序与参数一致
    success_count = sum(1 for r in results if r['status'] == 'ok')
    return {"total": len(image_urls), "success": success_count}

# Chord：并行处理 + 汇总回调
# 场景：批量获取多个商家的库存，汇总后生成补货建议
chord(
    (fetch_inventory.s(merchant_id) for merchant_id in merchant_ids),
    generate_restock_plan.s()  # 接收所有 fetch_inventory 的结果列表
).apply_async(countdown=300)

# Chain + Group + Chord 混合编排
# 场景：下载文件 → 并行处理多个 chunk → 合并结果 → 上传 CDN
complex_workflow = chain(
    download_file.s(file_url),
    group(process_chunk.s(i) for i in range(10)),
    merge_chunks.s(),
    upload_to_cdn.s()
)
# 注意：group 在 chain 中间时，group 的结果列表会作为 merge_chunks 的参数

# 更复杂的嵌套结构：先并行 A/B，等 A 完成后再串行 C/D
workflow = chord(
    group(task_a.s(), task_b.s()),
    chain(task_c.s(), task_d.s())
)

# celeryconfig.py
from kombu import Queue, Exchange

broker_url = 'amqp://user:password@rabbitmq:5672/myvhost'
result_backend = 'redis://redis:6379/0'

# 序列化：json 通用，msgpack 性能更好，pickle 功能最全但不安全
task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json', 'msgpack']

# Worker 配置
worker_prefetch_multiplier = 1      # 一次只取一个任务，避免任务被长时间占用
worker_max_tasks_per_child = 1000   # 执行 1000 个任务后重启 Worker，防止内存泄漏
worker_concurrency = 4              # 默认并发数，prefork 模式下为进程数

# 任务结果过期时间
result_expires = 3600               # 1 小时后自动清理结果
result_extended = True              # 存储更多元数据到结果后端

# 任务执行时间限制
task_time_limit = 3600              # 硬限制：1 小时 SIGKILL
task_soft_time_limit = 3300         # 软限制：55 分钟触发 SoftTimeLimitExceeded 异常

# 任务重试配置
task_default_retry_delay = 60
task_max_retries = 5

# 路由配置
task_routes = {
    'tasks.send_email': {'queue': 'email', 'routing_key': 'email.high'},
    'tasks.process_image': {'queue': 'image', 'routing_key': 'image.process'},
    'tasks.generate_report': {'queue': 'report', 'routing_key': 'report.daily'},
}

# 队列定义
task_queues = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('email', Exchange('email'), routing_key='email.#'),
    Queue('image', Exchange('image'), routing_key='image.#'),
    Queue('report', Exchange('report'), routing_key='report.#'),
    Queue('priority_high', Exchange('priority'), routing_key='priority.high'),
    Queue('priority_low', Exchange('priority'), routing_key='priority.low'),
)

# 默认队列
task_default_queue = 'default'
task_default_exchange = 'default'
task_default_routing_key = 'default'

# 推荐配置：Late ACK
task_acks_late = True
# 注意：Late ACK 下，任务必须在完成后才发送 ACK
# 如果 Worker 被 SIGKILL 或 OOM，任务会重新入队，由其他 Worker 执行

task_reject_on_worker_lost = True   # Worker 意外退出时拒绝任务（重新投递）

task_acks_on_failure_or_timeout = False  # 任务失败/超时不发送 ACK，触发重试

# Prefork（默认）：CPU 密集型任务如图片处理、数据计算
celery -A tasks worker -P prefork -c 4

# Gevent：IO 密集型，如大量 HTTP 请求
celery -A tasks worker -P gevent -c 1000

# Eventlet：旧版兼容
celery -A tasks worker -P eventlet -c 1000

# Solo：单线程调试
celery -A tasks worker -P solo

# Threads：Celery 5.2+ 原生线程池
celery -A tasks worker -P threads -c 50

from datetime import datetime, timedelta

# ETA：指定绝对执行时间
send_email.apply_async(args=['user@example.com'], eta=datetime(2024, 12, 25, 0, 0))

# Countdown：相对延迟（秒）
send_email.apply_async(args=['user@example.com'], countdown=3600)

# Expires：任务过期时间
send_email.apply_async(args=['user@example.com'], expires=datetime.now() + timedelta(hours=1))

# Priority：任务优先级（RabbitMQ/Kafka 支持）
send_email.apply_async(args=['user@example.com'], priority=0)   # 最高优先级

# Queue + Routing Key：精确路由
send_email.apply_async(args=['user@example.com'], queue='email', routing_key='email.urgent')

# 启动 Worker
# --queues 指定消费队列，-c 指定并发数
# -P prefork 默认多进程，-O fair 公平调度模式
celery -A tasks worker --loglevel=info \
    --queues=default,email,image \
    -c 4 \
    -O fair \
    --without-gossip \
    --without-mingle \
    --without-heartbeat

# 使用 Supervisor 管理 Worker
# /etc/supervisor/conf.d/celery.conf
[program:celery-worker]
command=/opt/venv/bin/celery -A tasks worker -c 4 -Q default,email
user=celery
directory=/opt/app
autostart=true
autorestart=true
redirect_stderr=true
stdout_logfile=/var/log/celery/worker.log

# 启动 Beat（定时任务调度）
celery -A tasks beat --loglevel=info \
    --schedule=/var/run/celery/celerybeat-schedule \
    --pidfile=/var/run/celery/beat.pid

# 启动多个 Worker（不同队列隔离）
celery -A tasks worker -Q default -n worker-default@%h -c 4
celery -A tasks worker -Q email -n worker-email@%h -c 8
celery -A tasks worker -Q image -n worker-image@%h -c 2

# Flower 监控界面
celery -A tasks flower \
    --port=5555 \
    --broker_api=redis://localhost:6379/0 \
    --persistent=True \
    --db=/var/lib/flower/flower.db
# 访问 http://localhost:5555 查看任务状态、Worker 负载、队列深度
# Flower 支持 REST API：curl http://localhost:5555/api/tasks

# 安装：pip install celery-prometheus-exporter
# 或使用 prometheus_client 自定义中间件

from prometheus_client import Counter, Histogram, Gauge, start_http_server
from celery.signals import task_prerun, task_postrun, task_retry, task_failure
import time

# 定义指标
task_counter = Counter('celery_task_total', 'Total tasks', ['task_name', 'status'])
task_latency = Histogram('celery_task_duration_seconds', 'Task duration', ['task_name'])
queue_length = Gauge('celery_queue_length', 'Queue length', ['queue_name'])
worker_count = Gauge('celery_worker_active', 'Active workers', ['queue_name'])

@task_prerun.connect
def task_started(sender=None, task_id=None, task=None, **kwargs):
    task._start_time = time.time()

@task_postrun.connect
def task_completed(sender=None, task_id=None, task=None, retval=None, state=None, **kwargs):
    duration = time.time() - getattr(task, '_start_time', time.time())
    task_latency.labels(task_name=task.name).observe(duration)
    task_counter.labels(task_name=task.name, status=state).inc()

@task_failure.connect
def task_failed(sender=None, task_id=None, exception=None, **kwargs):
    task_counter.labels(task_name=sender.name, status='failure').inc()

@task_retry.connect
def task_retried(sender=None, request=None, reason=None, **kwargs):
    task_counter.labels(task_name=sender.name, status='retry').inc()

# 启动指标服务器
start_http_server(9100)

import sentry_sdk
from sentry_sdk.integrations.celery import CeleryIntegration

# 初始化 Sentry，自动捕获 Celery 任务异常
sentry_sdk.init(
    dsn="https://xxx@yyy.ingest.sentry.io/zzz",
    integrations=[
        CeleryIntegration(
            propagate_traces=True,  # 传播 trace_id 关联分布式链路
        )
    ],
    traces_sample_rate=0.2,
)

# 在 Task 基类中添加上下文
class SentryTask(Task):
    abstract = True

    def apply_async(self, args=None, kwargs=None, **options):
        with sentry_sdk.push_scope() as scope:
            scope.set_tag("task_name", self.name)
            scope.set_context("celery", {"args": args, "kwargs": kwargs})
            return super().apply_async(args, kwargs, **options)

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        sentry_sdk.capture_exception(exc)
        super().on_failure(exc, task_id, args, kwargs, einfo)

@app.task(base=SentryTask)
def critical_task(data: dict):
    # 异常将自动上报到 Sentry，附带 task_id、args、kwargs
    risky_operation(data)

import logging
from pythonjsonlogger import jsonlogger

# 配置 JSON 格式日志
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
    '%(asctime)s %(levelname)s %(name)s %(message)s %(task_id)s %(task_name)s'
)
logHandler.setFormatter(formatter)
logger = logging.getLogger('celery')
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

# 在任务中注入上下文
@app.task(bind=True)
def logged_task(self, user_id: int):
    from celery._state import get_current_task
    logger.info(f"Processing for user {user_id}", extra={
        "task_id": self.request.id,
        "task_name": self.name,
        "user_id": user_id,
    })

# /etc/logstash/conf.d/celery.conf
input {
  file {
    path => "/var/log/celery/*.log"
    codec => json
  }
}
filter {
  mutate { add_field => { "service" => "celery-worker" } }
}
output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "celery-logs-%{+YYYY.MM.dd}"
  }
}

from kombu import Queue, Exchange

# RabbitMQ + DLX 配置
dlx_exchange = Exchange('dlx', type='direct')
dlq_queue = Queue('dead_letter', dlx_exchange, routing_key='dead_letter')

task_queues = (
    Queue('default', Exchange('default'), routing_key='default',
          queue_arguments={
              'x-dead-letter-exchange': 'dlx',
              'x-dead-letter-routing-key': 'dead_letter',
          }),
    dlq_queue,
)

from celery import Task
from celery.signals import task_failure
import redis

r = redis.Redis(host='localhost', port=6379, db=2)
DLQ_KEY = 'celery:dead_letter_queue'

class DLQTask(Task):
    abstract = True
    max_retries = 3

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        if self.request.retries >= self.max_retries:
            # 超过最大重试次数，写入死信队列
            dead_message = {
                'task_id': task_id,
                'task_name': self.name,
                'args': args,
                'kwargs': kwargs,
                'exception': str(exc),
                'traceback': str(einfo),
                'timestamp': datetime.utcnow().isoformat(),
            }
            r.lpush(DLQ_KEY, json.dumps(dead_message))
            logger.error(f"Task {task_id} moved to DLQ after {self.max_retries} retries")
        super().on_failure(exc, task_id, args, kwargs, einfo)

@app.task(base=DLQTask, bind=True, max_retries=3)
def process_payment(self, order_id: str, amount: float):
    try:
        charge(order_id, amount)
        return {"order_id": order_id, "status": "paid"}
    except PaymentError as exc:
        raise self.retry(exc=exc, countdown=60 * 2 ** self.request.retries)
    except NonRetryableError as exc:
        # 不可重试错误直接进 DLQ
        dead_message = { ... }
        r.lpush(DLQ_KEY, json.dumps(dead_message))
        raise

# 指数退避 + 最大延迟上限
@app.task(bind=True, max_retries=5, default_retry_delay=60)
def resilient_task(self, payload: dict):
    try:
        external_api_call(payload)
    except ExternalAPIError as exc:
        # 指数退避：60, 120, 240, 480, 960 秒
        countdown = min(60 * 2 ** self.request.retries, 3600)
        logger.warning(f"Retrying in {countdown}s (attempt {self.request.retries})")
        raise self.retry(exc=exc, countdown=countdown)
    except ValidationError as exc:
        # 业务错误不重试，直接失败
        logger.error(f"Business validation failed: {exc}")
        raise

# 带 Jitter 的退避策略（防止重试风暴）
import random

def jittered_backoff(retries: int, base: int = 60, max_delay: int = 3600) -> int:
    """带随机抖动的指数退避"""
    exp = min(base * 2 ** retries, max_delay)
    return random.randint(exp // 2, exp)

@app.task(bind=True, max_retries=5)
def jittered_task(self, data: dict):
    try:
        process(data)
    except Exception as exc:
        countdown = jittered_backoff(self.request.retries)
        raise self.retry(exc=exc, countdown=countdown)

# 批量重试死信队列任务
from celery import group

def retry_dlq_tasks(limit: int = 100):
    """从死信队列批量取出任务并重试"""
    tasks_to_retry = []
    for _ in range(limit):
        msg = r.rpop(DLQ_KEY)
        if not msg:
            break
        data = json.loads(msg)
        # 获取原始任务并重新投递
        task = app.tasks.get(data['task_name'])
        if task:
            tasks_to_retry.append(task.s(*data['args'], **data['kwargs']))

    if tasks_to_retry:
        group(tasks_to_retry).apply_async()
        logger.info(f"Retried {len(tasks_to_retry)} tasks from DLQ")

# app/celery_app.py
from celery import Celery

celery_app = Celery(
    'fastapi_tasks',
    broker='redis://redis:6379/0',
    backend='redis://redis:6379/1',
    include=['app.tasks.email', 'app.tasks.report']
)

celery_app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_expires=3600,
    task_track_started=True,
)

# app/main.py
from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import JSONResponse
from app.celery_app import celery_app
from app.tasks.email import send_email

app = FastAPI()

@app.post("/send-email")
async def enqueue_email(to: str, subject: str, body: str):
    """将邮件发送任务投递到 Celery"""
    task = send_email.delay(to, subject, body)
    return JSONResponse(
        status_code=202,
        content={"task_id": task.id, "status": "queued"}
    )

@app.get("/task/{task_id}")
async def get_task_status(task_id: str):
    """查询任务执行状态"""
    result = celery_app.AsyncResult(task_id)
    return {
        "task_id": task_id,
        "status": result.status,  # PENDING / STARTED / SUCCESS / FAILURE / RETRY
        "result": result.result if result.ready() else None,
        "traceback": result.traceback if result.failed() else None,
    }

# 轻量后台任务（FastAPI 原生，不经过 Celery）
@app.post("/light-task")
async def light_task(bg: BackgroundTasks):
    bg.add_task(simple_log, "User visited endpoint")
    return {"status": "background task queued"}

def simple_log(message: str):
    logger.info(message)

# app/scheduler.py
from datetime import datetime, timedelta
from typing import Optional
from celery import Celery

class TaskScheduler:
    """封装常用任务调度模式"""

    def __init__(self, celery_app: Celery):
        self.app = celery_app

    def schedule_once(
        self,
        task_name: str,
        args: tuple,
        countdown: Optional[int] = None,
        eta: Optional[datetime] = None,
        queue: str = 'default'
    ) -> str:
        """调度一次性任务"""
        task = self.app.send_task(
            task_name,
            args=args,
            countdown=countdown,
            eta=eta,
            queue=queue,
        )
        return task.id

    def schedule_periodic(
        self,
        task_name: str,
        schedule: int,  # 秒为单位
        args: tuple = (),
        queue: str = 'default'
    ):
        """动态添加定时任务（需配合 Beat / Redbeat）"""
        self.app.conf.beat_schedule[task_name] = {
            'task': task_name,
            'schedule': schedule,
            'args': args,
            'options': {'queue': queue},
        }

    def cancel_task(self, task_id: str) -> bool:
        """尝试撤销未执行的任务"""
        from celery.result import AsyncResult
        res = AsyncResult(task_id, app=self.app)
        res.revoke(terminate=True)
        return True

# 使用示例
scheduler = TaskScheduler(celery_app)

@app.post("/schedule-report")
async def schedule_report(user_id: int, run_at: datetime):
    task_id = scheduler.schedule_once(
        'app.tasks.report.generate_monthly_report',
        args=(user_id,),
        eta=run_at,
        queue='report'
    )
    return {"task_id": task_id, "scheduled_for": run_at.isoformat()}

from contextlib import contextmanager
from celery import Task
from sqlalchemy.orm import Session
from app.database import SessionLocal

class DBTask(Task):
    """自动注入数据库会话的 Task 基类"""
    abstract = True

    def __call__(self, *args, **kwargs):
        with get_db() as db:
            self._db = db
            return self.run(*args, **kwargs)

    @property
    def db(self) -> Session:
        return self._db

@contextmanager
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.task(base=DBTask, bind=True)
def update_user_stats(self, user_id: int):
    """自动拥有 db 会话的任务"""
    user = self.db.query(User).filter(User.id == user_id).first()
    user.stats_updated_at = datetime.utcnow()
    self.db.commit()
    return {"user_id": user_id, "updated": True}

# 默认值，适合短任务
worker_prefetch_multiplier = 4

# 推荐值，适合长任务或公平调度场景
worker_prefetch_multiplier = 1
# 配合 -O fair 启动参数，Worker 一次只取一个任务，
# 当前任务完成后再取下一个，避免任务堆积在某一 Worker

# CPU 密集型（图片处理、数据计算）：concurrency = CPU 核心数
celery -A tasks worker -P prefork -c $(nproc)

# IO 密集型（HTTP 请求、数据库查询）：concurrency 可以非常大
celery -A tasks worker -P gevent -c 1000

# 混合任务：按队列隔离，不同队列不同并发模型
celery -A tasks worker -Q cpu_tasks -P prefork -c 4
celery -A tasks worker -Q io_tasks -P gevent -c 500

# 策略 1：定期重启 Worker
worker_max_tasks_per_child = 1000   # 每个子进程处理 1000 个任务后重启
worker_max_memory_per_child = 200000  # 子进程使用超过 200MB 内存时重启（KB 为单位）

# 策略 2：在 Task 中手动释放大对象
@app.task
def process_large_file(file_path: str):
    import gc
    try:
        large_data = load_file(file_path)  # 可能占用大量内存
        result = transform(large_data)
        return result
    finally:
        del large_data  # 显式释放
        gc.collect()     # 强制垃圾回收

# 策略 3：使用 memory_profiler 检测
from memory_profiler import profile

@app.task
@profile
def suspected_leaky_task(data: list):
    # 运行后会在控制台输出逐行内存使用
    return heavy_operation(data)

# 使用 msgpack 替代 json，减少序列化开销
task_serializer = 'msgpack'
result_serializer = 'msgpack'
accept_content = ['json', 'msgpack']

# 关闭任务结果追踪（如果不需要）
task_ignore_result = True    # 全局关闭
# 或在单个任务上
task = my_task.s()
task.ignore_result = True

# 禁用 Rate Limit 检查（如果不需要）
task_annotations = {'*': {'rate_limit': None}}

# Broker connection pool 优化
broker_pool_limit = 10       # 连接池上限
broker_connection_timeout = 30
broker_heartbeat = 10        # 心跳间隔，及时发现死连接

# Worker 启动时预热（减少首次任务初始化时间）
worker_prefetch_multiplier = 1
worker_proc_alive_timeout = 4.0

# backend/websocket.py
import asyncio
from fastapi import WebSocket, WebSocketDisconnect
from celery.result import AsyncResult
from app.celery_app import celery_app

class ConnectionManager:
    def __init__(self):
        self.active_connections: dict[str, WebSocket] = {}

    async def connect(self, websocket: WebSocket, client_id: str):
        await websocket.accept()
        self.active_connections[client_id] = websocket

    def disconnect(self, client_id: str):
        self.active_connections.pop(client_id, None)

    async def send_progress(self, client_id: str, data: dict):
        if ws := self.active_connections.get(client_id):
            await ws.send_json(data)

manager = ConnectionManager()

@app.websocket("/ws/progress/{client_id}")
async def websocket_progress(websocket: WebSocket, client_id: str):
    await manager.connect(websocket, client_id)
    try:
        while True:
            data = await websocket.receive_json()
            task_id = data.get('task_id')
            if task_id:
                await monitor_task_progress(task_id, client_id)
    except WebSocketDisconnect:
        manager.disconnect(client_id)

async def monitor_task_progress(task_id: str, client_id: str):
    """轮询任务状态并通过 WebSocket 推送"""
    result = AsyncResult(task_id, app=celery_app)
    while not result.ready():
        await manager.send_progress(client_id, {
            "task_id": task_id,
            "status": result.status,
            "result": result.result if result.state == 'PROGRESS' else None,
        })
        await asyncio.sleep(1)

    # 最终状态
    await manager.send_progress(client_id, {
        "task_id": task_id,
        "status": result.status,
        "result": result.result if result.successful() else None,
        "error": result.traceback if result.failed() else None,
    })

from celery import shared_task, current_task
from celery.states import STARTED, SUCCESS, FAILURE

@shared_task(bind=True)
def long_running_task(self, items: list):
    """执行长时间任务并主动上报进度"""
    total = len(items)
    for i, item in enumerate(items):
        # 更新任务元数据中的进度
        self.update_state(
            state='PROGRESS',
            meta={
                'current': i + 1,
                'total': total,
                'percent': round((i + 1) / total * 100, 2),
                'current_item': str(item),
            }
        )
        process_item(item)

    return {'status': 'completed', 'processed': total}

# 前端获取进度的方式
# 方式 1：通过 Flower REST API
# GET http://flower:5555/api/task/info/<task_id>
# 返回包含 meta 字段的完整任务信息

# 方式 2：直接查询结果后端
result = long_running_task.AsyncResult(task_id)
if result.state == 'PROGRESS':
    meta = result.info
    print(f"Progress: {meta['percent']}% ({meta['current']}/{meta['total']})")

// frontend/progress.js
class TaskProgressTracker {
    constructor(wsUrl) {
        this.ws = new WebSocket(wsUrl);
        this.ws.onmessage = (event) => {
            const data = JSON.parse(event.data);
            this.updateUI(data);
        };
    }

    trackTask(taskId) {
        this.ws.send(JSON.stringify({ task_id: taskId }));
    }

    updateUI(data) {
        const { status, result, error } = data;
        const progressBar = document.getElementById('progress');

        if (status === 'PROGRESS' && result) {
            progressBar.style.width = `${result.percent}%`;
            progressBar.textContent = `${result.percent}%`;
        } else if (status === 'SUCCESS') {
            progressBar.style.width = '100%';
            progressBar.classList.add('bg-success');
        } else if (status === 'FAILURE') {
            progressBar.classList.add('bg-danger');
            console.error('Task failed:', error);
        }
    }
}

# Celery 5.x CLI 变化
# 旧写法（仍兼容）
celery worker -A proj

# 新推荐写法
celery -A proj worker

# 全局选项必须放在子命令之前
celery -A proj --quiet worker -l info   # OK
celery -A proj worker --quiet -l info   # 警告：全局选项位置

# Celery 5.x 使用 Click，命令分组更清晰

# Worker 管理
celery -A proj worker --loglevel=info -c 4

# Beat 定时任务
celery -A proj beat --loglevel=info

# 检查配置
celery -A proj config

# 查看注册任务列表
celery -A proj inspect registered

# 清空队列
celery -A proj purge

# 调用任务（命令行直接执行）
celery -A proj call tasks.send_email --args='["user@example.com","Hi","Hello"]'

# 1. 更新配置键名（全小写下划线）
# 4.x
task_serializer = 'json'
CELERY_TASK_SERIALIZER = 'json'  # 两种写法都支持

# 5.x（推荐）
task_serializer = 'json'
# CELERY_ 前缀写法仍支持但不再推荐

# 2. 移除 AMQP 结果后端使用
# 4.x
result_backend = 'amqp://guest:guest@localhost//'

# 5.x（必须更换）
result_backend = 'redis://localhost:6379/0'
# 或 result_backend = 'rpc://'（临时结果）

# 3. 序列化安全
# 4.x 默认 pickle，5.x 默认 json
task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']  # 为了安全，明确限制允许的序列化格式

# 4. 更新依赖
# requirements.txt
# celery>=5.0,<6.0
# kombu>=5.0
# billiard>=4.0

import celery

# 版本检测实现兼容性分支
CELERY_VERSION = tuple(map(int, celery.__version__.split('.')[:2]))

if CELERY_VERSION >= (5, 0):
    # Celery 5.x 配置
    app.conf.update(
        task_serializer='json',
        accept_content=['json'],
    )
else:
    # Celery 4.x 配置
    app.conf.update(
        task_serializer='pickle',
        accept_content=['pickle', 'json'],
    )

# Click 命令注册（Celery 5.x）
from celery import Celery
from celery.bin import worker as worker_cmd

app = Celery('compat_app')

# 自定义 CLI 命令（5.x Click 风格）
@app.on_after_configure.connect
def add_commands(sender, **kwargs):
    if CELERY_VERSION >= (5, 0):
        # 使用 Click 注册子命令
        pass  # 见 celery.bin.base 文档

# celeryconfig.py（Celery 5.x 推荐配置）
from kombu import Queue, Exchange

broker_url = 'redis://redis:6379/0'
result_backend = 'redis://redis:6379/1'

# 序列化
task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']

# 可靠性
task_acks_late = True
task_reject_on_worker_lost = True
worker_prefetch_multiplier = 1

# 性能
task_ignore_result = False       # 可根据场景全局关闭
result_expires = 7200            # 2 小时
worker_max_tasks_per_child = 1000

# 限流与安全
task_time_limit = 3600
task_soft_time_limit = 3300
task_default_retry_delay = 60
task_max_retries = 3

# 队列
task_queues = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('high', Exchange('tasks'), routing_key='tasks.high', queue_arguments={'x-max-priority': 10}),
)
task_default_queue = 'default'

# 监控
task_track_started = True        # 追踪任务 STARTED 状态