gpt4 book ai didi

python - 定期运行 Celery 任务(没有 Django)

转载 作者:太空宇宙 更新时间:2023-11-04 03:15:47 34 4
gpt4 key购买 nike

我正在尝试使用 Celery 定期运行一些功能(任务),比如每 3 秒一次。

我最接近的是只运行一次任务。

这是我的 Celery 配置文件:

# celeryconfig.py
from datetime import timedelta

BROKER_URL = 'amqp://guest@localhost//'

CELERY_RESULT_BACKEND = 'rpc://'

CELERYBEAT_SCHEDULE = {
'f1-every-3-seconds': {
'task': 'tasks.f1',
'schedule': timedelta(seconds=3),
'args': (1, 2)
},
'f2-every-3-seconds': {
'task': 'tasks.f2',
'schedule': timedelta(seconds=3),
'args': (3, 4)
},
}

这是我声明任务的地方:

# tasks.py:
import celeryconfig
from celery import Celery
from celery import task

dbwapp = Celery('tasks')
dbwapp.config_from_object(celeryconfig)

@dbwapp.task()
def f1(a, b):
print "F1: {0}, {1}".format(a, b)

@dbwapp.task()
def f2(a, b):
print "F2: {0}, {1}".format(a, b)

这是我的主程序运行的地方:

#tasks_runner.py:
from tasks import f1, f2, dbwapp


f1.delay(5, 6)
f2.delay(7, 8)

我用以下代码运行我的代码:python tasks_runner.py但没能使这两个函数定期运行。这是我得到的输出:

[2016-03-31 23:36:16,108: WARNING/Worker-9] F1: 5, 6
[2016-03-31 23:36:16,109: WARNING/Worker-6] F2: 7, 8

我做错了什么?如何让 f1 和 f2 定期运行?

最佳答案

您需要将 celery beat worker 作为一个单独的进程启动,而不是运行 task_runner.py:

celery -A proj beat

如此处所述:http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html

关于python - 定期运行 Celery 任务(没有 Django),我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/36344523/

34 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com