使用Flask动态调度Celery Beat任务

Question

我希望能让我的应用程序用户使用Celery beat启动/停止周期性的类似于crontab的任务。目前，我正在运行Celery。

venv/bin/celery worker -A celery_worker.celery --loglevel=info

我用这个简单的例子运行了Celery Beat：

@celery.task
def add(x, y):
    return x + y

在我的配置文件中：

CELERYBEAT_SCHEDULE = {
    'add-every-30-seconds': {
        'task': 'app.email.add',
        'schedule': timedelta(seconds=30),
        'args': (16, 16)
    },
}
CELERY_TIMEZONE = 'UTC'

然后我使用以下命令运行Celery定时任务工作者：

celery -A celery_worker.celery beat -s ~/Documents/cesco-automation/power/celerybeat-schedule

它完美地运行了！但我需要更多地控制时间表。

在我的应用程序壳上，我也能够做到这一点。

>>>add.apply_async([80,800],countdown=30)

>>> from datetime import datetime, timedelta     
>>> tomorrow = datetime.now() + timedelta(days=1)
>>> add.apply_async(args=[10, 10], eta=tomorrow)

这很好，但我正在开发一个家庭自动化应用程序，所以我还需要停止任务。我该怎么做？？

我还发现了这个链接，提到了django自定义调度程序类。这正是我所需要的。在Celery文档中，它提到了-S标志，但我不知道如何将类正确地添加到我的Flask应用程序中。我该如何在Flask中使用它？

我真的需要Celery Beat吗？除了crontab之外还有其他选择吗？Crontab似乎不够准确。

- CESCO

---

默认的celery调度器有点糟糕。更不用说处理时区的问题了。我会使用数据库调度器，可以看看django的那个http://celery.readthedocs.org/en/latest/userguide/periodic-tasks.html#using-custom-scheduler-classes - reptilicus

@CESCO，你好，你在Flask应用程序中是否尝试过使用Django调度程序？如果是，请分享一下。 - Saad Abdullah

好的，我发现了一些东西...而且它运行得很好...如果有人需要的话 https://github.com/tuomur/celery_sqlalchemy_scheduler - Saad Abdullah

@SaadAbdullah 你能给我展示一下吗？我一直得到 entry = self.Entry(**dict(b[key], name=key, app=self.app)) TypeError: string indices must be integers, not str。 - CESCO

经过一些调试，我意识到我没有正确地加载调度程序类。我尝试使用S标志和在我的配置文件中进行设置。你是如何做到这一点的？ - CESCO

Answer 1

这个功能将会在 Celery 4.0 版本中提供。目前只有开发版支持动态任务调度：http://docs.celeryproject.org/en/master/userguide/periodic-tasks.html#beat-entries

- Jakub Czaplicki

---

看起来Celery 4.0.0已于2016年11月4日发布（稳定版）：http://docs.celeryproject.org/en/latest/changelog.html#version-4-0-0。这意味着动态crontab周期性调度现在可用。 - Jakub Czaplicki

你有没有任何使用动态定时任务的示例？ - Deusdeorum

这仅适用于节拍开始之前。添加到计划中的任何内容都不会被捕获，直到节拍重新启动。仍然没有办法动态添加周期性任务。 - Dineshs91

@Dineshs91，我觉得这个评论和+2很令人困惑。我也在寻找动态的celery，然后在一年前的这个帖子评论中提到了来自Celery peeps的包django-celery-beat，它表明可以通过数据库更改重新加载计划列表。我想知道你具体指的是什么。 - xtian

Answer 2

尽管我们放弃了使用 Flask/Python 开发应用程序，但这个问题得到了解决，通过使用一个叫做 Kala 的调度器。它是一个简单的 JSON over HTTP API，受到 Chronus 的启发，与 Kala 相同，但由 Airbnb 提供了更为强大和可扩展的解决方案。它完全不受编程语言的限制。

- CESCO