Python中如何启动Celery并注册任务？

celery 启动过程中发现任务 有什么特殊的机制吗？ 同一个父目录下 1 2 3 个目录 1 2 里面的任务注册了 3 的任务就是注册不了。。


Python中如何启动Celery并注册任务？
1 回复

在Python中启动Celery并注册任务，核心是三步：创建Celery实例、定义任务函数、启动Worker。

1. 创建Celery实例 (celery_app.py) 首先安装Celery：pip install celery。然后创建一个Celery应用，指定消息代理（这里用Redis）和结果后端。

# celery_app.py
from celery import Celery

# 创建Celery实例，'my_project'是应用名
app = Celery('my_project',
             broker='redis://localhost:6379/0',  # 消息代理地址
             backend='redis://localhost:6379/0') # 结果后端地址

# 从配置文件加载配置（可选）
app.config_from_object('celery_config')

2. 定义任务 (tasks.py) 在同一个项目下创建任务模块。使用 @app.task 装饰器注册函数为异步任务。

# tasks.py
from celery_app import app

@app.task
def add(x, y):
    return x + y

@app.task
def multiply(x, y):
    return x * y

3. 启动Celery Worker 在终端切换到项目目录，运行以下命令启动Worker进程，让它监听任务队列：

celery -A celery_app worker --loglevel=info

这里的 -A celery_app 指定了Celery应用实例的位置（即 celery_app.py 中的 app 对象）。

运行任务 现在你可以在其他Python脚本或交互环境中调用这些异步任务了：

# 调用任务，这会立即返回一个AsyncResult对象，任务在后台执行
result = add.delay(4, 6)
print(result.get())  # 获取结果（会阻塞直到任务完成）

关键点总结

  • 消息代理：必须安装并运行一个（如Redis、RabbitMQ），Celery靠它传递任务消息。
  • 应用实例：所有任务模块需要能导入同一个 app 对象。
  • 自动发现：如果任务定义在其他模块，确保 celery_app.py 中配置了 app.autodiscover_tasks(['module_name']) 来自动发现。

一句话建议：确保你的消息代理服务（如Redis）已启动，否则Worker会连接失败。

回到顶部