【问题标题】:Get celery beat trigger time on task在任务中获取芹菜节拍触发时间
【发布时间】:2017-02-21 09:03:09
【问题描述】:

我正在尝试找到一种方法来获取触发 celery beat 以触发任务的时间条件。

获取datetime.now() 由于所有 celery 工作人员都在忙碌,因此通常会偏离任务按 celery beat 排队的时间。 例如:我将任务设置为每天 12:30 执行,但由于任务在 12:31 开始运行时所有工作人员都很忙。

无论任务执行的时间如何,我都需要知道哪个时间条件触发了任务。

编辑:

这就是我定义周期性任务的方式:

CELERYBEAT_SCHEDULE = {
'periodic_clear_task': {
    'task': 'app.tasks.periodic_clear_task',
    'schedule': crontab(hour=2),
    'args': ()
},
'periodic_update_task': {
    'task': 'app.tasks.periodic_update_task',
    'schedule': crontab(minute='00,30'),
    'args': ()
},
}

【问题讨论】:

  • 您可以在调用任务时将 datetime.now() 作为参数传递吗?
  • @JensAstrup 我该怎么做?编辑:我刚刚编辑了问题以包括我的定期任务的定义
  • 刚刚记住,一旦 celery worker 启动,调度程序就会被处理——而不是每次运行任务时。换句话说,将 datetime 作为参数传递只会给你启动 worker 的时间,这没有帮助。您正在寻找的内容可能在bound methods
  • 嗨莱昂内尔,你能得到这个答案吗?我也想这样做。
  • @MohammadMustaqeem 不幸的是没有。我最终将那些需要在他们自己的队列中更短的触发时间的任务排队,这样他们就不必等到其他任务完成并解决我的问题。这可能无法解决您的方案中的问题。它对我有用,因为这些任务并不多,可以移动到另一个队列,而不必担心顺序执行会丢失。如果您有任何其他问题,请随时提问。

标签: python django celery django-celery celerybeat


【解决方案1】:

我遇到了同样的问题,我想在周期性 celery 任务中使用任务触发时间而不是执行时间。我在具有任务触发时间的 celery 任务中找不到确切的字段。作为一种解决方法,我使用了 expires time of task。

task_expires_day =30

@app.on_after_configure.connect
def df_beat_setup(sender, **kwargs):
    pipeline_config = {}
    sender.add_periodic_task(
            crontab(
                    minute=0,
                    hour=8,
            ),
            df_scheduler_task.s(),
            args=(pipeline_config),
            name="df_trigger",
            queue="df_scheduler_queue",
            expires=60*60*24*task_expires_day,
            options={"time": datetime.now()}
    )


@app.task(name="df_scheduler_queue", bind=True, acks_late=True)
def df_scheduler_task(task: "celery.Task", pipeline_config: Dict, time: str) -> dict:
    task_trigger_time = parser.parse(task.request.expires)-timedelta(days=task_expires_day))
    
    ...

【讨论】:

    【解决方案2】:

    与此问题中的另一个 answer 类似,我已使用 expires(秒)来解决此问题。使用 Celery Beat 时间表时,可以指定如下:

    limit = 60
    
    CELERYBEAT_SCHEDULE = {
        ...,
    
        'periodic_update_task': {
            'task': 'app.tasks.periodic_update_task',
            'schedule': crontab(minute='00,30'),
            'args': (),
            'options': {'expires': 60 * 60 * 24 * limit }
        },
    }
    
    

    这里的限制很重要,因为现在我们分配了一个到期值,如果达到这个限制,任务就不会被执行。因此,它需要选择得足够大,以便根据需要处理任务。请记住,到期值以秒为单位指定。更多信息here

    当指定 expires 参数时,Celery 会自动将此秒值转换为要在内部使用的日期时间值。要访问函数中的过期时间,您必须绑定方法,然后从任务请求对象中检索。此任务对象应如下所示。

    <Context: {'lang': 'py', 'task': 'app.tasks.periodic_update_task', 'expires': '2021-10-23T21:11:26.031443+01:00', 'id': 'ef0e301a-b9b3-4930-8fa3-ca3fdf6b36cb', ... }>
    

    一旦有时间,您可以进行简单的计算以重新获得触发时间。

    @app.task(bind=True)
    def periodic_update_task(self):
        trigger_time = parser.parse(self.request.expires)-timedelta(days=limit))
    
    

    【讨论】:

      猜你喜欢
      • 2015-05-11
      • 2012-03-27
      • 2013-05-16
      • 2017-01-16
      • 2018-04-05
      • 1970-01-01
      • 1970-01-01
      • 2021-02-07
      • 1970-01-01
      相关资源
      最近更新 更多