如何使用Flask和Celery定期运行功能? [英] How to run a function periodically with Flask and Celery?

问题描述

我有一个烧瓶应用程序，大致如下所示:

I have a flask app that roughly looks like this:

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

   return "OK"

现在，我还想每10秒从该脚本运行一个函数.我不想为此睡觉.另外，我还有以下芹菜脚本:

Now in addition I would like to run a function every ten seconds from that script. I don't want to use sleep for that. I have the following celery script in addition:

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.我想念什么?

The script works fine, but the logger.info is not executed. What am I missing?

推荐答案

您是否正在运行Celery worker和Celery beat?计划的任务由 beat 处理，该任务在适当的时候将提到的任务排入队列.然后， Worker 会计算数字并执行您的任务.

Do you have Celery worker and Celery beat running? Scheduled tasks are handled by beat, which queues the task mentioned when appropriate. Worker then actually crunches the numbers and executes your task.

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

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

Your task however looks like it's calling the Flask app's logger. When using the worker, you probably don't have the Flask application around (since it's in another process). Try using a normal Python logger for the demo task.

这篇关于如何使用Flask和Celery定期运行功能?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！