python - Celery 任务相同的时间表但有偏移

Question

我有几项任务，如下所示:

CELERYBEAT_SCHEDULE = {
    'task1': {
        'task': 'api.tasks.task1',
        'schedule': timedelta(seconds=10),
    },
    'task2': {
        'task': 'api.tasks.task2',
        'schedule': timedelta(seconds=30),
    },
    'task3': {
        'task': 'api.tasks.task3',
        'schedule': timedelta(seconds=15),
    },
    ...
}

因此，task1 将在 *:*:10、*:*:20、*:*:30、*:*:40、*:*:50 和 *:*:00 中运行

task2 将在 *:*:30 和 *:*:00 中运行

task3 将在 *:*:15、*:*:30、*:*:45 和 *:*:00 中运行

然后任务总是在 *:*:30 和 *:*:00 中并发。有什么办法可以添加偏移量吗？我想要得到这样的东西:

task1 (offset=2) 在 *:*:12、*:*:22、*:*:32、*:*:42、*:*:52 和 *:*:02 中运行

task2 (offset=7) 在 *:*:37 和 *:*:07 中运行

task3 (offset=0) 在 *:*:15、*:*:30、*:*:45 和 *:*:00 中运行

我已经阅读了文档，我想我必须使用 crontab，但是没有其他更好的方法吗？并且 crontab 没有秒配置:-(

Answer 1

根据celery documentation :

You can also define your own custom schedule types, by extending the interface of schedule.

这是我的解决方案:

from datetime import timedelta

from celery import Celery
from celery.schedules import schedule


class MySchedule(schedule):
    def __init__(self, run_every=None, offset=None):
        self._run_every = run_every
        self._offset = offset if offset is not None else timedelta(seconds=0)
        self._do_offset = True if self._offset else False
        super(MySchedule, self).__init__(
            run_every=self._run_every + self._offset)

    def is_due(self, last_run_at):
        ret = super(MySchedule, self).is_due(last_run_at)
        if self._do_offset and ret.is_due:
            self._do_offset = False
            **self._offset = datetime.timedelta(seconds=0)** #bug fix
            self.run_every = self._run_every
            ret = super(MySchedule, self).is_due(last_run_at)
        return ret

    def __reduce__(self):
        return self.__class__, (self._run_every, self._offset)


app = Celery('tasks', broker='pyamqp://guest@localhost//')

app.conf.beat_schedule = {
    'task1': {
        'task': 'tasks.task1',
        'schedule': MySchedule(
            run_every=timedelta(seconds=10), offset=timedelta(seconds=2)),
    },
    'task2': {
        'task': 'tasks.task2',
        'schedule': MySchedule(
            run_every=timedelta(seconds=30), offset=timedelta(seconds=7)),
    },
    'task3': {
        'task': 'tasks.task3',
        'schedule': MySchedule(
            run_every=timedelta(seconds=15), offset=timedelta(seconds=0)),
    },
}


@app.task
def task1():
    print('task1')


@app.task
def task2():
    print('task2')


@app.task
def task3():
    print('task3')

您可以编写自己的 MySchedule 并将其从 BaseSchedule 扩展以拥有更多控制权。