### 三级标题：在单个 Python 文件中配置 Celery 的生产者、任务和设置以实现异步任务执行 在单个 Python 文件中实现 Celery 的完整配置，包括生产者、任务定义和基本设置，是简化异步任务处理的一种有效方式。以下是一个完整的示例，展示如何在一个文件中定义 Celery 用例，使其能够异步执行任务。 #### 配置 Celery 实例 首先，需要创建一个 Celery 实例，并指定 Broker 和 Backend 的地址。通常使用 Redis 或 RabbitMQ 作为 Broker。例如，使用 Redis 作为消息中间件时，可以如下定义： ```python from celery import Celery app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0') ``` 该配置创建了一个名为 `tasks` 的 Celery 实例，并指定 Redis 作为 Broker 和 Backend [^4]。 #### 定义异步任务 接下来，可以在同一个文件中定义一个或多个异步任务。使用 `@app.task` 装饰器将函数标记为 Celery 任务。例如，定义一个简单的加法任务： ```python @app.task def add(x, y): return x + y ``` 该任务可以在异步环境中被调用并执行，适用于处理耗时较长的操作，如图像处理、发送邮件等 [^4]。 #### 启动 Celery Worker 定义完任务后，需要启动 Celery Worker 来监听任务队列并执行任务。可以通过命令行启动 Worker： ```bash celery -A tasks worker --loglevel=info ``` 其中 `tasks` 是包含 Celery 实例的模块名称。如果文件名为 `tasks.py`，则可以直接使用该命令启动 Worker [^4]。 #### 提交异步任务 最后，可以在另一个 Python 脚本或同一文件中调用任务。例如，使用 `delay()` 方法异步执行任务： ```python result = add.delay(4, 6) print(result.get(timeout=10)) ``` 该代码将任务 `add(4, 6)` 提交到 Broker，并等待执行结果。使用 `get()` 方法可以获取任务的返回值 [^4]。 #### 完整示例代码 以下是一个完整的示例代码，展示了如何在一个 Python 文件中定义 Celery 应用、任务和执行异步任务： ```python from celery import Celery # 创建 Celery 应用并配置 Broker 和 Backend app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0') # 定义异步任务 @app.task def add(x, y): return x + y # 提交异步任务（可选） if __name__ == '__main__': result = add.delay(4, 6) print(result.get(timeout=10)) ``` 在运行该文件之前，确保已启动 Redis 服务和 Celery Worker，以便任务可以被正确调度和执行 [^4]。 ### 三级标题：总结 通过上述方式，可以在一个 Python 文件中完成 Celery 的完整配置，包括定义生产者、任务和异步执行流程。该方法适用于简单的异步任务处理场景，如定时任务、消息推送、数据处理等 [^3]。确保 Broker 和 Backend 服务正常运行，并正确配置 Celery 实例，以避免任务执行失败。