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

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

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

   return "OK"

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

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 未执行。我缺少什么？

你有 Celery 工作程序和 Celery 击败运行吗？计划任务由以下人员处理beat，它会在适当的时候对提到的任务进行排队。Worker然后实际处理数字并执行你的任务。

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

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