gpt4 book ai didi

celery - 用 Chronos 替换 Celerybeat

转载 作者:行者123 更新时间:2023-12-04 04:41:37 25 4
gpt4 key购买 nike

成熟到什么程度 Chronos ?它是像 celery-beat 这样的调度程序的可行替代方案吗?

现在我们的调度实现了一个周期性的“心跳”任务,它检查“未完成”的事件并在它们过期时触发它们。我们正在使用 python-dateutil定义这个的规则。

我们正在寻找这种方法的替代方案,并且 Chronos似乎是一个非常有吸引力的替代方案:1)它可以减少使用心跳计划任务的必要性,2)它支持以 RESTful 提交 ISO8601 格式的事件,3)有一个有用的管理界面,以及 4)它可以扩展。

关键的要求是调度需要可从 Web 界面即时配置。这就是为什么不能开箱即用地使用 celerybeat 的内置调度。

我们是否要通过切换到 Chronos 来踢自己的脚?

最佳答案

This SO有解决您的动态周期性任务问题的方法。目前这不是公认的答案:

 from djcelery.models import PeriodicTask, IntervalSchedule
from datetime import datetime

class 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()

我没用过 djcelery但是,据说它有一个用于动态周期性任务的管理界面。

关于celery - 用 Chronos 替换 Celerybeat,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/17098377/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com