flask+celery的代码例子

下面是一个简单的 Flask Celery 代码例子：

# app.py
from flask import Flask
from celery import Celery

app = Flask(__name__)
app.config.update(
    CELERY_BROKER_URL='redis://localhost:6379',
    CELERY_RESULT_BACKEND='redis://localhost:6379'
)

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

@app.route('/')
def index():
    task = celery.send_task('tasks.add', args=[2, 3])
    result = task.get()
    return f'The result is {result}'

if __name__ == '__main__':
    app.run()

# tasks.py
from app import celery

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

这个例子中，我们创建了一个简单的 Flask 应用，在其中定义了一个路由 `/`，当访问该路由时，会调用 Celery 中的任务 `tasks.add`，传递给它两个参数 `2` 和 `3`，并返回它的执行结果。

在 `app.py` 中，我们首先设置了 Celery 的配置项，然后创建了一个 Celery 实例，并传入了 Flask 应用的名称和 Redis 作为消息代理和结果后端。然后我们更新了 Celery 的配置项，并将其与 Flask 应用的配置项合并。

然后我们在路由函数中使用 `celery.send_task` 方法调用 `tasks.add` 任务，并传递参数。我们使用 `task.get()` 方法来获取任务的执行结果，并将其返回给客户端。

在 `tasks.py` 中，我们定义了一个简单的任务 `add`，它接收两个参数并返回它们的和。使用 `@celery.task` 装饰器来将该函数转换为 Celery 任务。