如何向Celery动态添加/删除定期任务（celerybeat）

Question

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

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)

我还想用（pseudocode）之类的东西动态地停止或删除该任务：

celery.beat.remove_task(some_unique_task_id)

要么

celery.beat.stop(some_unique_task_id)

仅供参考，我没有使用djcelery，它使您可以通过django管理员管理定期任务

Answer 1

这个问题在google网上论坛上得到了回答。我不是作者，所有功劳归Jean Jean这是一个适当的解决方案。确认可以正常工作，在我的场景中，我将“定期任务”子类化，并根据它创建了一个模型，因为我可以根据需要向模型中添加其他字段，也可以添加“终止”方法。您必须将定期任务的enabled属性设置为False并保存，然后再将其删除。整个子类不是必须的，schedule_every方法是真正起作用的方法。当您准备终止任务时（如果您没有将其子类化），您可以使用PeriodicTask.objects.filter（name = ...）搜索任务，将其禁用，然后将其删除。希望这可以帮助！from djcelery.models import PeriodicTask, IntervalSchedulefrom datetime import datetimeclass TaskScheduler(models.Model):    periodic_task = models.ForeignKey(PeriodicTask)    @staticmethod    def schedule_every(task_name, period, every, args=None, kwargs=None):    """ schedules a task by name every "every" "period". So an example call would be:         TaskScheduler('mycustomtask', 'seconds', 30, [1,2,3])          that would schedule your custom task to run every 30 seconds with the arguments 1,2 and 3 passed to the actual task.     """        permissible_periods = ['days', 'hours', 'minutes', 'seconds']        if period not in permissible_periods:            raise Exception('Invalid period specified')        # create the periodic task and the interval        ptask_name = "%s_%s" % (task_name, datetime.datetime.now()) # create some name for the period task        interval_schedules = IntervalSchedule.objects.filter(period=period, every=every)        if interval_schedules: # just check if interval schedules exist like that already and reuse em            interval_schedule = interval_schedules[0]        else: # create a brand new interval schedule            interval_schedule = IntervalSchedule()            interval_schedule.every = every # should check to make sure this is a positive int            interval_schedule.period = period             interval_schedule.save()        ptask = PeriodicTask(name=ptask_name, task=task_name, interval=interval_schedule)        if args:            ptask.args = args        if kwargs:            ptask.kwargs = kwargs        ptask.save()        return TaskScheduler.objects.create(periodic_task=ptask)    def stop(self):        """pauses the task"""        ptask = self.periodic_task        ptask.enabled = False        ptask.save()    def start(self):        """starts the task"""        ptask = self.periodic_task        ptask.enabled = True        ptask.save()    def terminate(self):        self.stop()        ptask = self.periodic_task        self.delete()        ptask.delete()

Answer 2

有一个名为django-celery-beat的库，它提供了所需的模型。为了使其能够动态加载新的定期任务，必须创建自己的Scheduler。from django_celery_beat.schedulers import DatabaseSchedulerclass AutoUpdateScheduler(DatabaseScheduler):    def tick(self, *args, **kwargs):        if self.schedule_changed():            print('resetting heap')            self.sync()            self._heap = None            new_schedule = self.all_as_schedule()            if new_schedule:                to_add = new_schedule.keys() - self.schedule.keys()                to_remove = self.schedule.keys() - new_schedule.keys()                for key in to_add:                    self.schedule[key] = new_schedule[key]                for key in to_remove:                    del self.schedule[key]        super(AutoUpdateScheduler, self).tick(*args, **kwargs)    @property    def schedule(self):        if not self._initial_read and not self._schedule:            self._initial_read = True            self._schedule = self.all_as_schedule()        return self._s