时间戳

首页 > 解决方案 > 在计划任务的情况下如何从芹菜中获取任务ID（节拍）

python - 在计划任务的情况下如何从芹菜中获取任务ID（节拍）

问题描述

要访问 celery 任务的信息，我需要 task_id。当手动启动 celery 任务时，我可以使用 task.id 轻松获取此任务的 id（并将其写入 DB 或执行其他操作）。如果我使用 celery-beat，它会定期向工作人员发送任务，这似乎是不可能的。

所以我的问题是，如何在 beat 将任务发送给 celery 的工人的那一刻从任务中获取 id？

在工作人员收到任务的那一刻，控制台会显示任务 ID。所以我担心的是，在任务被beat发送给工人的那一刻，它没有任务ID。

手动获取task_id的案例：

task = tasks.LongRunningTask.delay(username_from_formTargetsLaden, password_from_formTargetsLaden, url_from_formTargetsLaden)

task_id = task.id

也许你们中的一些人有想法？

标签： pythonrabbitmqcelerycelerybeat

解决方案

我找到了这个小问题的答案：

如果您需要最初由 beat 发送的任务的任务 ID，您可以简单地向您的（计划的）工作任务添加一个检查功能。

配置定期任务

这是每天上午 11:08（UTC）“提醒”芹菜开始任务的时间表。

@celery.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    test = sender.add_periodic_task(crontab(minute=8, hour=11), CheckLists.s(app.config['USR'], app.config['PWD']))

定期执行的任务

这是工作人员收到beat的“提醒”后​​，celery将执行的计划任务。

@celery.task(bind=True)
def CheckLists(self, arg1, arg2):
    #get task_id von scheduled Task 'Check-List'
    i = inspect()
    activetasks = i.active()
    list_of_tasks = {'activetasks': activetasks}
    task_id = list_of_tasks['activetasks']['celery@DESKTOP-XXXXX'][0]['id']   #adapt this section depending on environment (local, webserver, etc...)
    task_type = "CHECK_LISTS"
    task_id_to_db = Tasks(task_id, task_type)
    db.session.add(task_id_to_db)
    db.session.commit()

    long_runnning_task 
    [...more task relevant code here...]

所以我正在利用app.control.inspect它让你检查正在运行的工人。它在引擎盖下使用远程控制命令。你会得到i.active()一本字典，你可以很容易地解析它。

只要我没有找到任何文档如何更容易地从定期任务中获取 task_id，我就会坚持使用该解决方案。

保存任务 ID 后，您可以通过 AJAX 轻松轮询任务状态等。

希望对你们有帮助:)

推荐阅读