本文实例为大家分享了celery动态设置定时任务的具体代码，供大家参考，具体内容如下
首先celery是一种异步任务队列，如果还不熟悉这个开源软件的请先看看官方文档，快速入门。
这里讲的动态设置定时任务的方法不使用数据库保存定时任务的信息，所以是项目重启后定时任务配置就会丢失，如果想保存成永久配置，可以考虑保存到数据库、redis或者使用pickle、json保存成文件，在项目启动时自动载入。
方法原理介绍
先来看一下celery的beat运行过程。
上图是beat的主要组成结构，beat中包含了一个service对象，service中包含了一个scheduler对象，scheduler中包含了一个schedule字典，schedule中key对应的的value才是真正的定时任务，是整个beat中最小的单元。
首先分别介绍一下各个对象和它们运行的过程，beat是celery.apps.beat.Beat类创建的对象，调用beat.run()方法就可以启动beat,下面是beat.run()方法的源码。
def run(self):
print(str(self.colored.cyan(
'celery beat v{0} is starting.'.format(VERSION_BANNER))))
self.init_loader()
self.set_process_title()
self.start_scheduler()
重点是在run()方法里调用了start_scheduler()方法，而start_scheduler()方法本质上是创建了一个service对象（celery.beat.Service类），并调用service.start()方法，下面是beat.start_scheduler()方法的源码。
def start_scheduler(self):
if self.pidfile:
platforms.create_pidlock(self.pidfile)
service = self.Service(
app=self.app,
max_interval=self.max_interval,
scheduler_cls=self.scheduler_cls,
schedule_filename=self.schedule,
)
print(self.banner(service))
self.setup_logging()
if self.socket_timeout:
logger.debug('Setting default socket timeout to %r',
self.socket_timeout)
socket.setdefaulttimeout(self.socket_timeout)
try:
self.install_sync_handler(service)
service.start()
except Exception as exc:
logger.critical('beat raised exception %s: %r',
exc.__class__, exc,
exc_info=True)
raise
调用了service.start()之后，会进入一个死循环，先使用self.scheduler.tick()获取下一个任务a的定时点到现在时间的间隔，然后进入睡眠，睡眠结束之后判断如果self.scheduler里的下一个任务a可以执行，就立即执行，并获取self.scheduler里的下下一个任务b的定时点到现在时间的间隔，进入下一次循环。下面是service.start()的源码。
def start(self, embedded_process=False):
info('beat: Starting...')
debug('beat: Ticking with max interval->%s',
humanize_seconds(self.scheduler.max_interval))
signals.beat_init.send(sender=self)
if embedded_process:
signals.beat_embedded_init.send(sender=self)
platforms.set_process_title('celery beat')
try:
while not self._is_shutdown.is_set():
interval = self.scheduler.tick()
if interval and interval > 0.0:
debug('beat: Waking up %s<p>本文来源gao!daima.com搞$代!码#网#</p>.',
humanize_seconds(interval, prefix='in '))
time.sleep(interval)
if self.scheduler.should_sync():
self.scheduler._do_sync()
except (KeyboardInterrupt, SystemExit):
self._is_shutdown.set()
finally:
self.sync()
service.scheduler默认是celery.beat.PersistentScheduler类的实例对象，而celery.beat.PersistentScheduler其实是celery.beat.Scheduler的子类，所以scheduler.schedule是celery.beat.Scheduler类中的字典，保存的是celery.beat.ScheduleEntry类型的对象。ScheduleEntry的实例对象保存了定时任务的名称、参数、定时信息、过期时间等信息。celery.beat.Scheduler类实现了对schedule的更新方法即update_from_dict(self, dict_)方法。下面是update_from_dict(self, dict_)方法的源码。
def _maybe_entry(self, name, entry):
if isinstance(entry, self.Entry):
entry.app = self.app
return entry
return self.Entry(**dict(entry, name=name, app=self.app))
def update_from_dict(self, dict_):
self.schedule.update({
name: self._maybe_entry(name, entry)
for name, entry in items(dict_)
})
转载请注明原文链接：celery实现动态设置定时任务
