Celery 任务队列集成

Litefs 提供了与 Celery 的无缝集成，简化异步任务处理。

安装

pip install celery redis

快速开始

1. 初始化 Celery

from litefs.core import Litefs
from litefs.tasks import Celery

app = Litefs(
    host='0.0.0.0',
    port=8080,
    celery_broker='redis://localhost:6379/0',
    celery_backend='redis://localhost:6379/1',
)

celery = Celery(app)

2. 定义任务

@celery.task
def send_email(to, subject, body):
    # 发送邮件
    return {'status': 'sent', 'to': to}

3. 异步执行

# 提交任务
result = send_email.delay('user@example.com', 'Hello', 'World')
task_id = result.id

# 查询状态
status = celery.get_task_status(task_id)

# 获取结果
result = celery.get_task_result(task_id)

任务选项

基本选项

@celery.task(
    name='custom_task',        # 任务名称
    queue='high_priority',     # 队列名称
    max_retries=3,             # 最大重试次数
    default_retry_delay=60,    # 重试延迟（秒）
    time_limit=300,            # 硬时间限制
    soft_time_limit=270,       # 软时间限制
)
def process_data(data):
    return {'processed': data}

绑定任务

@celery.task(bind=True)
def process_file(self, filepath):
    try:
        # 处理文件
        return {'status': 'success'}
    except Exception as exc:
        raise self.retry(exc=exc)

定时任务

from celery.schedules import crontab

@celery.task
def cleanup_sessions():
    # 清理会话
    pass

# 每 30 分钟执行
celery.schedule(
    name='cleanup-every-30-minutes',
    task=cleanup_sessions,
    schedule=30 * 60,
)

# 每天凌晨 2 点执行
celery.schedule(
    name='daily-report',
    task=send_daily_report,
    schedule=crontab(hour=2, minute=0),
)

API 参考

Celery 类

方法	说明
`task(func, **options)`	任务装饰器
`schedule(name, task, schedule, args, kwargs)`	配置定时任务
`get_task(task_id)`	获取任务结果对象
`get_task_status(task_id)`	获取任务状态
`get_task_result(task_id, timeout)`	获取任务结果
`revoke_task(task_id, terminate)`	撤销任务
`get_active_tasks()`	获取活跃任务列表
`get_registered_tasks()`	获取已注册任务列表
`get_queue_length(queue)`	获取队列长度
`health_check()`	健康检查

配置选项

配置项	默认值	说明
`celery_broker`	`None`	Broker URL
`celery_backend`	`None`	Result Backend URL
`celery_config`	`None`	额外配置字典

启动 Worker

# 启动 Worker
celery -A app.celery worker --loglevel=info

# 启动 Beat（定时任务）
celery -A app.celery beat --loglevel=info

# 同时启动 Worker 和 Beat
celery -A app.celery worker --beat --loglevel=info

完整示例

参见 examples/05-production/celery_tasks.py，其中定义了 process_task、send_email、cleanup_cache 三个任务与 Beat 定时调度配置。