Celery动态子任务的同步等待机制：突破链式限制

import time
from typing import List
from celery import Celery, Task, AsyncResult
from celery.signals import task_postrun

# 假设的Celery应用实例
app = Celery('my_app', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

# 模拟的JobMaster和常量，用于日志记录
class JobMaster:
    def get_job(self, job_id, job_title):
        print(f"[JobMaster] Getting job {job_id} - {job_title}")
        return self, job_id
    def log_message(self, log_message, status=None, job_score=None):
        print(f"[JobMaster] Log: {log_message} (Status: {status}, Score: {job_score})")

class Consts:
    IN_PROGRESS = "IN_PROGRESS"
    COMPLETED = "COMPLETED"
    ERRORS_FOUND = "ERRORS_FOUND"

# 模拟的子任务
@app.task
def task_dummy_subtask(parent_task_name: str, job_id: int = None):
    job, _ = JobMaster().get_job(job_id, job_title="dummy subtask")
    job.log_message(log_message=f"Subtask {parent_task_name} started.")
    time.sleep(2) # 模拟耗时操作
    job.log_message(log_message=f"Subtask {parent_task_name} finished successfully.")
    return f"Result from {parent_task_name}"

# 模拟的中间函数，用于创建子任务
def intermediary_dummy_subtask_function(parent_task_name: str, job_id: int = None) -> AsyncResult:
    job, _ = JobMaster().get_job(job_id, job_title="dummy task")
    job.log_message(log_message=f"Intermediary function for {parent_task_name} has been reached, will now make a task")
    r = task_dummy_subtask.apply_async(kwargs={"parent_task_name": parent_task_name, "job_id": job_id},
                                       add_to_parent=True)
    return r

@app.task(bind=True)
def task_dummy_task1(self: Task, part_number: int, job_id: int = None):
    job, job_id = JobMaster().get_job(job_id, job_title="dummy task")
    sleeping_duration = 1
    subtask_ids = []
    job.log_message(log_message=f"Entered dummy task 1 with sleeping duration of {sleeping_duration}")

    # 直接创建子任务
    job.log_message(log_message="In dummy task1, creating subtask a")
    subtask = task_dummy_subtask.apply_async(kwargs={"parent_task_name": "task1_a", "job_id": job_id},
                                             add_to_parent=True)
    subtask_ids.append(subtask.id)

    job.log_message(log_message="In dummy task1, creating subtask b")
    subtask = task_dummy_subtask.apply_async(kwargs={"parent_task_name": "task1_b", "job_id": job_id},
                                             add_to_parent=True)
    subtask_ids.append(subtask.id)

    # 通过中间函数创建子任务
    job.log_message(log_message="In dummy task1, creating intermediary subtask c")
    subtask = intermediary_dummy_subtask_function(parent_task_name="task1_c", job_id=job_id)
    subtask_ids.append(subtask.id)

    time.sleep(sleeping_duration) # 模拟主任务的其他操作

    # 等待所有子任务完成
    wait_for_tasks_to_complete(async_ids=subtask_ids, job_id=job_id,
                                    msg="Waiting in dummy task1 for subtasks to complete")

    job.log_message(log_message="Finished dummy task1 main body")
    return part_number

def wait_for_tasks_to_complete(async_ids: List[str], job_id: int = None, msg: str = None, timeout: int = 300):
    job, _ = JobMaster().get_job(job_id, job_title="waiting for refresh data")

    # 复制一份ID列表，因为在循环中会移除已完成的任务
    remaining_async_ids = list(async_ids) 

    job.log_message(log_message=f"Waiting for {len(remaining_async_ids)} tasks to complete, {msg}", status=Consts.IN_PROGRESS, job_score=0)
    job.log_message(log_message=f"tasks: {remaining_async_ids}", status=Consts.IN_PROGRESS, job_score=0)

    count_down = timeout
    while count_down > 0:
        # 遍历剩余任务，检查其状态
        tasks_to_check = list(remaining_async_ids) # 避免在迭代时修改列表
        all_succeeded_in_this_check = True

        for async_id in tasks_to_check:
            result = app.AsyncResult(async_id) # 获取任务结果对象
            status = result.status

            if status == "SUCCESS":
                returned_value = result.result
                job.log_message(log_message=f"Task {async_id} confirmed status SUCCESS with {returned_value=}")
                remaining_async_ids.remove(async_id) # 从待检查列表中移除
            elif status in ["FAILURE", "REVOKED", "RETRY"]: # 考虑失败或撤销状态
                job.log_message(log_message=f"Task {async_id} failed or revoked with status {status}. Aborting wait.", status=Consts.ERRORS_FOUND)
                # 根据业务需求，可以选择在此处抛出异常或返回失败
                return False 
            else:
                all_succeeded_in_this_check = False # 仍有任务未完成或未成功

        # 如果所有任务都已完成
        if not remaining_async_ids:
            job.log_message(log_message="Finished waiting for refresh data, all tasks succeeded",
                            status=Consts.COMPLETED, job_score=100)
            return True # 所有任务成功完成

        count_down -= 1
        job.log_message(log_message=f"There are {len(remaining_async_ids)} tasks remaining. Waiting...", status=Consts.IN_PROGRESS)
        time.sleep(1) # 避免忙等，每秒检查一次

    # 超时退出
    job.log_message(log_message=f"After waiting for {timeout=} seconds, some tasks did not complete on time. Remaining tasks: {remaining_async_ids}", 
                    status=Consts.ERRORS_FOUND, job_score=100)
    return False # 超时，未所有任务完成

# task_dummy_task1 的最后部分
# ...
    time.sleep(sleeping_duration) 

    # 等待所有子任务完成
    if wait_for_tasks_to_complete(async_ids=subtask_ids, job_id=job_id,
                                    msg="Waiting in dummy task1 for subtasks to complete"):
        job.log_message(log_message="Finished dummy task1 main body after all subtasks completed.")
    else:
        job.log_message(log_message="Dummy task1 finished with issues: subtasks did not complete on time or failed.", status=Consts.ERRORS_FOUND)

    return part_number