标签: celerybeat

是否可以配置芹菜节拍的复杂时间表？例如,像这样:

在周一到周五,用参数(x1,y1)做工作A,然后做工作B在星期六,星期日,用参数(x2,y2)做工作A,不做工作B

我知道我可以实现一个高频率的"滴答"任务来检查这个时间表,但是如果已经存在的话,我不想重新发明轮子.

所以我正在使用虚拟机（vagrant）进行开发，当我在其中启动 celerybeat 时，我收到此消息：

[2014-07-15 10:16:49,627: INFO/MainProcess] beat: Starting...
[W 140715 09:16:51 state:74] Substantial drift from celery@worker_publications may mean clocks are out of sync.  Current drift is
    3600 seconds.  [orig: 2014-07-15 09:16:51.476125 recv: 2014-07-15 10:16:51.474109]

[W 140715 09:16:51 state:74] Substantial drift from celery@worker_queue may mean clocks are out of sync.  Current drift is
    3600 seconds.  [orig: 2014-07-15 09:16:51.480642 recv: 2014-07-15 10:16:51.475021]

当我在里面约会时，我得到了一个Tue Jul 15 09:25:11 UTC 2014，但问题是我住在葡萄牙，我的主机给了我Ter Jul 15 10:25:39 WEST 2014。

对我来说解决这个问题的最佳方法是什么？

当我把这个上线的时候呢？

我正在使用celery 3.1.12 …

我有一个任务calculate_common_locations,它运行一次CELERYBEAT_SCHEDULE.该任务只是调用数据库中的函数:

@app.task
def calculate_common_locations():
    db.execute("SELECT * FROM calculate_centroids('b')")

这是条目CELERYBEAT_SCHEDULE:

CELERYBEAT_SCHEDULE = {
   'common_locations': {
        'task': 'clients.tasks.calculate_common_locations',
        'schedule': crontab(hour=23, day_of_week='sun'), #every week
    },
    [..]
}

该计划包括每天运行一次或每10秒运行一次的更多任务.这些任务似乎没有多次重新运行.芹菜花显示任务执行超过20次.第一个按计划启动,运行约100秒,成功然后再次启动.

只有一个celerybeat运行:

ps -Af | grep celerybeat 
foo     24359   779  0 01:53 ?        00:00:04 [celeryd: celery@celery:MainProcess] -active- (worker --beat --app=cloud.celeryapp:app --concurrency=10 -l INFO -s /home/foo/run/celerybeat-schedule --pidfile=/home/foo/run/celerybeat.pid)         

这就是芹菜的开始方式(通过supervisord):

celery worker --beat --app=cloud.celery app:app --concurrency=10 -l INFO -s /home/foo/run/celerybeat-schedule --pidfile=/home/foo/run/celerybeat.pid

我没有使用--concurrency = 10 …

我正在使用Celery 3.1.23版，将动态计划的任务动态添加到芹菜节拍中。我有一名芹菜工作者和一名芹菜节拍实例正在运行。

运行task.delay（）可以触发标准的芹菜任务。当我将计划的定期任务定义为配置中的设置时，芹菜节拍将其运行。

但是，我需要的是能够添加在运行时在指定的crontab上运行的任务。将任务添加到持久性调度程序后，芹菜节拍似乎无法检测到新添加的新任务。我可以看到celery-schedule文件确实有一个包含新任务的条目。

码：

scheduler = PersistentScheduler(app=current_app, schedule_filename='celerybeat-schedule')
scheduler.add(name="adder",
          task="app.tasks.add",
          schedule=crontab(minute='*/1'),
          args=(1,2))
scheduler.close()

当我跑步时：

print(scheduler.schedule)

我得到：

{'celery.backend_cleanup': <Entry: celery.backend_cleanup celery.backend_cleanup() <crontab: 0 4 * * * (m/h/d/dM/MY)>,
'adder': <Entry: adder app.tasks.add(1, 2) <crontab: */1 * * * * (m/h/d/dM/MY)>}

？

请注意，app.tasks.add具有@celery.task装饰器。

我有一个我在Heroku上启动的应用程序,但是当我启动服务器时,Celery节拍过程无法启动.

Procfile

web: gunicorn -w 4 connect.wsgi
celery: python manage.py celeryd -c 3 --beat

在Heroku应用程序启动后,可以看到工作人员已启动:

$ heroku ps

=== web (Free): gunicorn -w 4 connect.wsgi (1)
web.1: starting 2016/07/13 16:17:18 -0400 (~ 9s ago)

=== celery (Free): python manage.py celeryd -c 3 --beat (1)
celery.1: up 2016/07/13 16:17:25 -0400 (~ 2s ago)

但是,为了让Celery击败进程运行,我必须在Heroku中明确地启动它:

heroku run python manage.py celerybeat

Celery beat在当地推出.这是Heroku的限制还是我做错了什么？

*使用celery 3.1.25，因为django-celery-beat 1.0.1在安排定期任务方面存在问题。

最近，我遇到了一个关于celerybeat的问题，该问题导致调度程序似乎忘记了间隔一天或更长的定期任务。如果我更改时间间隔，every 5 seconds 则任务将正常执行（每5秒），并且last_run_at属性会更新。这意味着celerybeat在一定程度上响应了调度程序，但是如果我重置last_run_atie PeriodicTask.objects.update(last_run_at=None)，则every day不再有运行间隔的任务。

Celerybeat在某一时刻崩溃，并且可能损坏了某些内容，因此我创建了一个新的virtualenv和数据库，以查看问题是否仍然存在。我想知道是否有一种方法可以检索下一次运行时间，这样我就不必等待一天就可以知道我的定期任务是否已执行。

我也尝试过使用，inspect <active/scheduled/reserved>但全部归还empty。使用djcelery的数据库调度程序执行定期任务是否正常？

这是安排任务的功能：

def schedule_data_collection(request, project):
    if (request.method == 'POST'):
        interval = request.POST.get('interval')

        target_project = Project.objects.get(url_path=project)    
        interval_schedule = dict(every=json.loads(interval), period='days')

        schedule, created = IntervalSchedule.objects.get_or_create(
            every=interval_schedule['every'],
            period=interval_schedule['period'],
        )

        task_name = '{} data collection'.format(target_project.name)

        try:
            task = PeriodicTask.objects.get(name=task_name)
        except PeriodicTask.DoesNotExist:
            task = PeriodicTask.objects.create(
                interval=schedule,
                name=task_name,
                task='myapp.tasks.collect_tool_data',
                args=json.dumps([target_project.url_path])
            )
        else:
            if task.interval != schedule:
                task.interval = schedule

            if task.enabled is False: …

我编写了一个模块,根据项目设置中的字典列表(通过导入django.conf.settings)动态添加定期芹菜任务.我这样做是使用一个函数add_tasks来调度一个函数,该函数使用uuid设置中给出的特定函数进行调用:

def add_tasks(celery):
    for new_task in settings.NEW_TASKS:
        celery.add_periodic_task(
            new_task['interval'],
            my_task.s(new_task['uuid']),
            name='My Task %s' % new_task['uuid'],
        )

像这里建议我使用on_after_configure.connect信号来调用我的函数celery.py:

app = Celery('my_app')

@app.on_after_configure.connect
def setup_periodic_tasks(celery, **kwargs):
    from add_tasks_module import add_tasks
    add_tasks(celery)

这个设置适用于两者celery beat,celery worker但在我用于uwsgi服务我的django应用程序的地方中断了我的设置.Uwsgi运行顺利,直到视图代码第一次使用celery的.delay()方法发送任务.在这一点上,似乎芹菜被初始化,uwsgi但在上面的代码中永远阻止.如果我从命令行手动运行它然后在它阻塞时中断,我得到以下(缩短的)堆栈跟踪:

Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/kombu/utils/objects.py", line 42, in __get__
    return obj.__dict__[self.__name__]
KeyError: 'tasks'

During handling of the above exception, another exception occurred: …

我的烧瓶应用程序由四个容器组成：Web应用程序，postgres，rabbitMQ和Celery。由于我有定期运行的芹菜任务，因此我使用的是芹菜节拍。我已经像这样配置了docker-compose文件：

version: '2'
services:
  rabbit:
    # ...      
  web:
    # ...
  rabbit:
    # ...
  celery:
    build:
        context: .
        dockerfile: Dockerfile.celery

我的Dockerfile.celery看起来像这样：

# ...code up here...
CMD ["celery", "-A", "app.tasks.celery", "worker", "-B", "-l", "INFO"]

当我在文档中阅读到我不应该使用该-B选项进行生产时，无论如何我还是匆忙添加了该选项（并忘记了对其进行更改），并很快了解到我的计划任务正在运行多次。对于那些感兴趣的人，如果您ps aux | grep celery在celery容器中执行a操作，则会看到多个celery + beat进程正在运行（但是应该只有一个Beat进程，但是应该有许多worker进程）。从文档中我不确定为什么不应该-B在生产环境中运行，但是现在我知道了。

所以然后我将Dockerfile.celery更改为：

# ...code up here...
CMD ["celery", "-A", "app.tasks.celery", "worker", "-l", "INFO"]
CMD ["celery", "-A", "app.tasks.celery", "beat", "-l", "INFO"]

不，当我启动我的应用程序时，工作进程启动，但节拍没有启动。当我翻转这些命令时，将首先调用beat，然后启动beat，但不会启动工作进程。所以我的问题是：如何在容器中运行芹菜工作者+一起殴打？我已经梳理了许多文章/文档，但仍然无法弄清楚。

已编辑

我将Dockerfile.celery更改为以下内容：

ENTRYPOINT [ "/bin/sh" ]
CMD [ "./docker.celery.sh" ]    

我的docker.celery.sh文件如下所示：

#!/bin/sh -ex
celery -A …

目前，我在 Azure 实例上使用作业计划程序设置了定期任务。这些在固定时间触发 API (Django) 端点。

我想让这些时间变得动态（这不适用于此解决方案）。计划是直接从 Django 触发这些任务。计划时间将存储在我的数据库 (MySQL) 中并被检索以创建计划作业。当这些值改变时，调度程序也应该相应地改变。

在查看了 Celery 后，似乎使用定期任务 crontab 计划可以工作。使用这个，是否可以根据数据库中的值设置我的计划时间？

看来我还需要一个 Redis 实例。由于我只会使用 Celery 来执行定期任务，这仍然是正确的方法吗？

我正在使用 docker compose 和 django-postgres-rabbitmq

一切都很顺利，即使 django 容器已启动并运行到我想要的程度，makemigrations它会随着 django-celery-beat 停止

Migrations for 'django_celery_beat':
django_1        |   /usr/local/lib/python2.7/site-packages/django_celery_beat/migrations/0005_auto_20190512_1531.py
django_1        |     - Alter field event on solarschedule
django_1        | Traceback (most recent call last):
django_1        |   File "/app/manage.py", line 10, in <module>
django_1        |     execute_from_command_line(sys.argv)
django_1        |   File "/usr/local/lib/python2.7/site-packages/django/core/management/__init__.py", line 364, in execute_from_command_line
django_1        |     utility.execute()
django_1        |   File "/usr/local/lib/python2.7/site-packages/django/core/management/__init__.py", line 356, in execute
django_1        |     self.fetch_command(subcommand).run_from_argv(self.argv)
django_1        |   File "/usr/local/lib/python2.7/site-packages/django/core/management/base.py", line 283, in run_from_argv
django_1        |     self.execute(*args, **cmd_options)
django_1        | …