使用celery canvas中的group任务类型实现分组任务

「爱情、让人受尽委屈。」 2023-02-11 08:49 8阅读 0赞

**1. 需求背景：**

>     某个业务需求实现过程需要对发布到celery的异步任务进行拆分，得到两个并行执行的子任务，同时在主任务的进度反馈中需要更新子任务的进度；

**2. 实现方案：**

> **   ** 1. 任务发布端通过 "send\_task()" 方式进行任务发布；
> 
>     2. worker接收到任务后，创建 group 任务，子任务的调用方式为 "task signature";
> 
>     3. 进度的更新利用了 celery res.get() 方法中的 "on\_message" 回调进行任务状态跟踪；

**3. 实施细节：**

> 出于对业务内容的保密考虑，以demo形式展示实施细节；

** 3.1 group\_task.py**

import logging
    
    from celery import group
    from celery.result import allow_join_result
    from celery_once import QueueOnce
    
    from celery_app import app
    from .task_A import task_A
    from .task_B import task_B
    
    
    # on_message回调值传递组任务内的任务进度, 更新组任务的进度需要获取group task对象
    task_obj = None
    # on_message回调每次只会返回一个子任务的任务进度, 做个全局进度保存
    task_status = dict(
        task_A=dict(status="PROGRESS", progress=0, msg=""),
        task_B=dict(status="PROGRESS", progress=0, msg="")
    )
    
    def progress_update(meta_data):
        """
        更新进度的回调函数
        :param meta_data: 回调会自动获取任务反馈的进度meta_data
        :return: None
        """
        global task_obj
        global task_status
        task_status[meta_data["result"]["task_name"]]["status"] = meta_data.get("status")
        if task_status["task_A"]["status"] == "SUCCESS" and task_status["task_B"]["status"] == "SUCCESS":
            group_status = "SUCCESS"
        elif task_status["task_A"]["status"] == "FAILED" or task_status["task_B"]["status"] == "FAILED":
            group_status = "FAILED"
        else:
            group_status = "PROGRESS"
        if meta_data.get("status") == "PROGRESS":
            task_status[meta_data["result"]["task_name"]]["progress"] = meta_data["result"]["progress"]
        if task_obj:
            logging.info("progress update, task_status: {}".format(task_status))
            task_obj.update_state(state=group_status, meta_data=dict(**task_status))
        else:
            logging.error("can't get global task_obj!")
    
    @app.task(bind=True, base=QueueOnce)
    def group_task(self, arg_1, arg_2):
        """
        - 将被拆分的组任务，拆成以下两个子任务：
          - task_A
          - task_B
        """
        # 更新全局变量task_obj为当前组任务
        global task_obj
        task_obj = self
        group_task_id = self.request.id
        # 创建组任务
        res = group(task_A.s(group_task_id, arg_1, arg_2, arg_3),
                    task_B.s(group_task_id, arg_1, arg_2, arg_3))()
        with allow_join_result():
            result = res.get(on_message=progress_update, propagate=False)
        logging.info("group task finish.")

**3.2 task\_A.py**

import logging
    
    from celery_once import QueueOnce
    
    from celery_app import app
    
    
    @app.task(bind=True, base=QueueOnce)
    def task_A(self, arg_1, arg_2, arg_3):
        # 更新进度
        self.update_state(state="PROGRESS",
                          meta=dict(
                              task_name="task_A",
                              progress=0,
                              status="progress"
                          ))
        # demo这里模拟做任务
        n = 0.1
        for i in range(10):
            ratio = n * i
            self.update_state(state="PROGRESS",
                              meta=dict(
                                  task_name="task_A",
                                  progress=ratio,
                                  status="progress"
                              ))
        # 任务完毕
        self.update_state(state="PROGRESS",
                          meta=dict(
                              task_name="task_A",
                              progress=1,
                              status="finish"
                          ))

**tips:**

**    这里有一个需要注意的地方就是组任务内使用 on\_message 回调获取组内任务的进度信息, 需要用到celery提供的上下文管理器** allow\_join\_result **，否则会引发异常：**

RuntimeError: Never call result.get() within a task!