标签: celerybeat

如果我有一个定义如下的函数:

def add(x,y):
  return x+y

有没有办法动态添加这个函数作为芹菜PeriodicTask并在运行时启动它？我希望能够做类似(伪代码)的事情:

some_unique_task_id = celery.beat.schedule_task(add, run_every=crontab(minute="*/30"))
celery.beat.start(some_unique_task_id)

我还想用类似(伪代码)的东西动态地停止或删除该任务:

celery.beat.remove_task(some_unique_task_id)

要么

celery.beat.stop(some_unique_task_id)

仅供参考我没有使用djcelery,它允许您通过django管理员管理定期任务.

Django的事情变化太大,所以我不能使用3.1.我需要一些帮助.

我读到了在django中创建任务,并阅读了Periodic Tasks文档.但我不知道如何在django中定期执行任务.我想这是因为我的低级英语..

在旧版本的Celery中,我导入了djcelery&crontab并在settings.py中设置了CELERYBEAT_SCHEDULE,并由manage.py执行.

但似乎我再也不能以这种方式执行芹菜deamon了.比我应该把CELERYBEAT_SCHEDULE放在哪里？在docs的django示例中,它们设置os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings') 在proj/proj/celery.py中.(1)所以在settings.py中设置celerybeat (就像之前一样)可以吗？

(2)如果可以的话,在django中制作周期性任务的一个小例子将非常有帮助.当我阅读教程时,最令人困惑的部分是文件路径.如果你不想提供一个完整的例子,如果你解释我应该在哪里制作任务,设置节拍和执行deamon,我将非常感激.

谢谢阅读.

celeryd不需要pid文件,但celerybeat似乎.有没有办法禁用它？我正在使用Upstart来管理进程,因此使用pidfile是多余的.

我有一个check_orders任务,定期执行.它创建了一组任务,以便我可以计算执行任务所花费的时间,并在完成任务时执行某些操作(这是res.join [1]和grouped_subs的目的)分组的任务是成对的链式任务.

我想要的是当第一个任务不满足条件(失败)时不执行链中的第二个任务.对于我的生活,我无法理解这一点,我觉得这对于作业队列管理器来说是非常基本的功能.当我尝试在[2]之后注释掉的东西(引发异常,删除回调)......由于某种原因,我们在check_orders中遇到了join()(它打破了组).我已经尝试将ignore_result设置为False以及所有这些任务,但它仍然不起作用.

@task(ignore_result=True)
def check_orders():
    # check all the orders and send out appropriate notifications
    grouped_subs = []

    for thingy in things:
       ...

        grouped_subs.append(chain(is_room_open.subtask((args_sub_1, )), 
                        notify.subtask((args_sub_2, ), immutable=True)))

    res = group(grouped_subs).apply_async()

    res.join()         #[1]
    logger.info('Done checking orders at %s' % current_task.request.id))

@task(ignore_result=True)
def is_room_open(args_sub_1):
    #something time consuming
    if http_req_and_parse(args_sub_1):
        # go on and do the notify task
        return True
    else:
        # [2]
        # STOP THE CHAIN SOMEHOW! Don't execute the rest of the chain, how?
        # None of the following …

我正在寻找推荐的解决方案来解决celerybeat是芹菜/ rabbitmq部署的单点故障.通过搜索网络,到目前为止我找不到任何有意义的东西.

在我的情况下,每天定时调度员开始一系列可能运行半天或更长时间的工作.由于只能有一个celerybeat实例,如果它发生了某些事情或它正在运行的服务器,将无法运行关键作业.

我希望已经有一个可行的解决方案,因为我不能成为唯一需要可靠(集群或类似)调度程序的人.如果我不需要,我不想诉诸某种数据库支持的调度程序.

你如何诊断为什么manage.py celerybeat不执行任何任务？

我通过supervisord运行celerybeat命令:

/usr/local/myapp/src/manage.py celerybeat --schedule=/tmp/celerybeat-schedule-myapp --pidfile=/tmp/celerybeat-myapp.pid --loglevel=INFO

Supervisord似乎运行celerybeat就好了,日志文件显示:

[2013-06-12 13:17:12,540: INFO/MainProcess] Celerybeat: Starting...
[2013-06-12 13:17:12,571: WARNING/MainProcess] Reset: Account for new __version__ field
[2013-06-12 13:17:12,571: WARNING/MainProcess] Reset: Account for new tz field
[2013-06-12 13:17:12,572: WARNING/MainProcess] Reset: Account for new utc_enabled field

我有几个定期任务显示为启用http://localhost:8000/admin/djcelery/periodictask,应该每隔几分钟运行一次.但是,celerybeat日志从不显示正在执行的任何内容.为什么会这样？

我的celerybeat.conf

[program:celerybeat]
command=/path/app/env/bin/celery beat -A project.tasks --loglevel=INFO
environment=PYTHONPATH=/path/app/env/bin

user=nobody
numprocs=1
stdout_logfile=/var/log/celeryd.log
stderr_logfile=/var/log/celeryd.log
autostart=true
autorestart=true
startsecs=10
stopwaitsecs = 600
killasgroup=true
priority=998

当我启动主管时,我收到一个错误:

pidfile_fd = os.open(self.path, PIDFILE_FLAGS, PIDFILE_MODE)
celery.platforms.LockFailed: [Errno 13] Permission denied: '/celerybeat.pid'

不知道怎么解决这个问题？

在我的Django项目中,我使用Celery和Rabbitmq在后台运行任务.我正在使用芹菜节拍调度程序来执行定期任务.如何以编程方式检查芹菜节拍是否正常运行？

多天后,我有一个工作芹菜和芹菜节拍任务列表,结果使用django_celery_results存储.但是,当我查看表记录时,它没有任何有用的信息.

是否可以将任务ID设置为人类可读的东西？

一个例子是使用demo任务,它返回no,但是一个不可读的任务id

tasks.py

@app.task
def test(a,b):
    return a + b

app.settings中的调度程序

CELERYBEAT_SCHEDULE = {
    'test_task': {
        'task': 'home.tasks.test',
        'schedule': crontab(minute='*/1'),
    },

我一直试图设置django + celery + redis + celery_beats,但它给了我麻烦.文档非常简单,但是当我运行django服务器,redis,芹菜和芹菜节拍时,没有任何内容被打印或记录(我所有的测试任务都会记录它的东西).

这是我的文件夹结构:

- aenima 
 - aenima
   - __init__.py
   - celery.py

 - criptoball
   - tasks.py

celery.py看起来像这样:

from __future__ import absolute_import, unicode_literals
import os
from django.conf import settings
from celery import Celery


# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'aenima.settings')

app = Celery("criptoball")
app.conf.broker_url = 'redis://localhost:6379/0'

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related …