### Celery Async 安装与使用教程 Celery 是一个强大的分布式任务队列，支持异步任务处理。为了在 Celery 中使用 `async` 和 `await` 功能，需要进行一些特定的配置和操作。 #### 1. 安装 Celery 及相关依赖 首先，确保安装了最新版本的 Celery 和 Python 的异步支持库。可以通过以下命令安装所需依赖： ```bash pip install celery[redis] aiohttp ``` 这里假设使用 Redis 作为消息代理[^3]。如果使用其他消息代理（如 RabbitMQ），可以替换为相应的依赖。 #### 2. 配置 Celery 应用 在项目的 `__init__.py` 文件中创建 Celery 实例并加载配置模块： ```python # __init__.py from celery import Celery app = Celery('demo') # 创建 Celery 实例 app.config_from_object('celery_app.celeryconfig') # 加载配置模块 ``` 在 `celery_app/celeryconfig.py` 文件中定义 Celery 的配置： ```python # celery_app/celeryconfig.py broker_url = 'redis://localhost:6379/0' # 消息代理地址 result_backend = 'redis://localhost:6379/0' # 结果存储地址 task_serializer = 'json' # 任务序列化格式 result_serializer = 'json' # 结果序列化格式 accept_content = ['json'] # 接受的内容类型 timezone = 'UTC' # 时区设置 enable_utc = True # 启用 UTC 时间 ``` #### 3. 在 Celery 中运行异步任务 Celery 默认不支持直接运行 `async def` 定义的函数，因此需要通过 `asyncio.run()` 或类似方式包装异步函数。以下是一个示例： ```python import asyncio from celery import Celery app = Celery('demo') app.config_from_object('celery_app.celeryconfig') def run_async(coro): # 包装异步函数 return asyncio.run(coro) @app.task def do_sth(*args, **kwargs): # Celery 任务 return run_async(_do_sth(*args, **kwargs)) async def _do_sth(*args, **kwargs): # 异步任务逻辑 await sub_task() return 'done.' async def sub_task(): # 子异步任务 await asyncio.sleep(1) # 模拟 IO 操作 print("子任务完成") ``` #### 4. 调用异步任务 通过以下代码调用异步任务，并返回任务 ID： ```python def demo_to_call_task(): args, kwargs = (), {} task = do_sth.delay(*args, **kwargs) # 异步执行任务 return {'task_id': task.id} ``` #### 5. 启动 Celery 工作进程 在终端中启动 Celery 工作进程以处理任务： ```bash celery -A demo worker --loglevel=info ``` #### 6. 监控任务状态 可以通过任务 ID 查询任务状态： ```python from demo import app task_id = "your_task_id_here" task = app.AsyncResult(task_id) if task.ready(): result = task.get() print(f"任务结果: {result}") else: print("任务尚未完成") ``` ### 注意事项 - 每个 Celery 任务中只能运行一次 `asyncio.run()`，否则可能导致事件循环冲突[^2]。 - 确保所有依赖库支持异步操作，例如使用 `aioredis` 替代同步的 `redis` 库[^2]。 ---