如何使用 Flask 和 Celery 定期运行一个函数？

Question

我有一个 Flask 应用程序，大致如下所示：

app = Flask(__name__)
@app.route('/',methods=['POST'])
def foo():
   data = json.loads(request.data)
   # do some stuff

   return "OK"

现在我还想运行从该脚本每 10 秒执行一个函数。我不想为此使用睡眠。我还有以下芹菜脚本：

from celery import Celery
from datetime import timedelta
celery = Celery('__name__')

CELERYBEAT_SCHEDULE = {
    'add-every-30-seconds': {
        'task': 'tasks.add',
        'schedule': timedelta(seconds=10)
    },
}



@celery.task(name='tasks.add')
def hello():
    app.logger.info('run my function')

脚本工作正常，但 logger.info 没有执行。我错过了什么？

Answer 1

你有 Celery worker 和 Celery 打败运行吗？计划任务由 beat 处理，它会在适当的时候将提到的任务排队。 Worker 然后实际计算数字并执行您的任务。

celery worker --app myproject--loglevel=info
celery beat --app myproject

然而，您的任务看起来像是在调用 Flask 应用程序的记录器。使用 worker 时，您可能没有 Flask 应用程序（因为它在另一个进程中）。尝试为演示任务使用普通的 Python 记录器。

Answer 2

默认情况下，celery 任务将运行在 Flask 应用程序上下文之外，因此它无法访问 Flask 应用程序实例。但是，通过使用 Flask 应用程序对象的 app_context 方法，在运行执行任务时创建 Flask 应用程序上下文非常容易。

app = Flask(__name__)
celery = Celery(app.name)

@celery.task
def task():
    with app.app_context():
        app.logger.info('running my task')

This article 作者 Miguel Grinberg 是一个很好的地方，可以让你了解在 Flask 应用程序中使用 Celery 的基础知识。

Answer 3

好吧，celery beat 也可以嵌入到常规的 celery worker 中，在命令中使用 -B 参数。

celery -A --app myproject --loglevel=info -B

仅推荐用于开发环境。对于生产，您应该运行像 documentation 提到的那样分别击败和芹菜工人。否则，您的周期性任务将运行不止一次。

如何使用 Flask 和 Celery 定期运行一个函数？

How to run a function periodically with Flask and Celery?

python

celery

flask

如何使用 Flask 和 Celery 定期 运行 一个函数？

How to run a function periodically with Flask and Celery?

python

celery

flask

如何使用 Flask 和 Celery 定期运行一个函数？