Python异步/线程实现自监控最优方案？多MainTask并行执行指导

环境 #

Python 3.10.6

背景 #

ManagerTask类负责执行Task方法，这些Task实际是异步Celery任务。已实现任务执行追踪功能，但不确定当前asyncio用法是否规范（了解Flower工具）。目前程序仅支持单个Main()编排Task执行，实际需要扩展为多MainTask会话并行执行。

现有代码 #

ManagerTask类代码 #

class ManagerTask:
    def __init__(self, id:int ) -> None:
        self.id = id
        self.tasks: List["Task"] = []
        self.executed_tasks: List["Task"] = []
        self.state = State.SCHEDULED

    def load_tasks(self, configs: List[Dict]):
        # 加载任务的代码

    async def run(self):
        """ 执行所有任务并启动结果监控 """
        execute_task = asyncio.create_task(self.task_exec())
        monitor_progresss = asyncio.create_task(self.monitor_progress())
        await execute_task
        await monitor_progresss
        
    async def monitor_progress(self):
        """ 若任一任务失败，标记Main.state为失败；全部成功则标记为成功 """
        import time
        failure_flag = False
        count_success = 0
        i = 0

        while True:     
            state = self.executed_tasks[i].task_run.state 

            if state == 'SUCCESS':
                count_success += 1      
                i += 1

            if state == 'FAILURE':
                failure_flag = True
                i += 1

            print(f'Rule_ID:{self.executed_tasks[i].rule_id} \
                 \nCelery_UUID:{self.executed_tasks[i].task_celery_id} \
                 \nstatus - {self.executed_tasks[i].task_run.state}')
                 
            # 所有任务已处理（成功/失败）
            if i == len(self.executed_tasks) -1:
                if failure_flag:
                    self.state = State.FAILED
                elif count_success == len(self.executed_tasks) :
                    self.state = State.FINISHED
                break

            # 否则等待
            await asyncio.sleep(3)                        
                
         
    async def task_exec(self):
        for task in self.starting_tasks:
            task.run()     

客户端执行代码 #

# ...其他代码
cust = MainTask(id=1)    
cust.load_tasks(configs=rules, db_connections=db_connections)
asyncio.run(cust.run())

print("MainTask State:" + cust.state)

输出示例 #

Rule_ID:4
Celery_UUID:fecde27c-b58a-43cd-9498-3478404c248b                    
status - FAILURE
....
Rule_ID:6
Celery_UUID:85df9bba-3d75-4b00-a533-a81cd3f6afb3                    
status - SUCCESS

MainTask State:Failed

并行代码尝试 #

async def exec_id(id):
    cust = MainTask(id=id)    
    cust.load_tasks(configs=rules ,...)
    await cust.run()

async def main():
    ids = [111,222,333]

    for id in ids:
        await exec_id(id)

asyncio.run(main())

核心问题 #

1. 当前的asyncio实现方式是否正确？
2. 运行多MainTask时，应采用线程还是asyncio？因程序通过Celery执行任务，倾向用asyncio但不确定，同时希望得到现有并行代码方案的指导。

1. 当前asyncio实现的问题 #

你的asyncio用法存在几个不规范和逻辑问题：

• 索引越界风险：monitor_progress中获取任务状态后直接i +=1，后续打印时访问self.executed_tasks[i]会出现索引越界（比如i已是最后一个元素索引时，i+1超出列表范围）。
• 时序逻辑漏洞：task_exec中的task.run()如果是提交Celery任务（通常同步调用但立即返回AsyncResult），execute_task会很快执行完毕，此时executed_tasks可能还未被完整填充，monitor_progress直接访问会报错。
• 无意义的async包装：task_exec里只是循环提交Celery任务，属于IO操作但Celery已处理异步逻辑，这里用async函数包装没有实际意义，改成普通同步方法即可。
• 监控逻辑低效不严谨：固定3秒轮询不够灵活，且未处理任务处于PENDING等中间状态的情况，会导致i无法递增，陷入无限循环。

优化建议 #

• 将task_exec改为同步方法，提交所有Celery任务后，把任务结果对象存入executed_tasks。
• 监控逻辑改为遍历每个任务，等待其进入最终状态（成功/失败），避免索引操作：

async def monitor_progress(self):
    failure_flag = False
    count_success = 0
    for task in self.executed_tasks:
        while True:
            state = task.task_run.state
            if state in ('SUCCESS', 'FAILURE'):
                break
            await asyncio.sleep(1)
        
        print(f'Rule_ID:{task.rule_id} \
             \nCelery_UUID:{task.task_celery_id} \
             \nstatus - {state}')
        
        if state == 'SUCCESS':
            count_success +=1
        else:
            failure_flag = True
    
    if failure_flag:
        self.state = State.FAILED
    elif count_success == len(self.executed_tasks):
        self.state = State.FINISHED

2. 多MainTask并行的方案选择 #

优先用asyncio实现并行 #

Celery任务的提交和状态查询都是IO密集型操作，asyncio能在单线程内高效处理多个MainTask的并行调度，避免线程切换开销，完全适配你的场景。

现有并行代码的问题 #

你当前的代码是串行执行：for id in ids: await exec_id(id)会逐个等待每个exec_id完成后才执行下一个，没有任何并行效果。

正确的asyncio并行实现 #

使用asyncio.gather同时运行多个协程：

async def exec_id(id):
    cust = MainTask(id=id)    
    cust.load_tasks(configs=rules ,...)
    await cust.run()
    return cust.state

async def main():
    ids = [111,222,333]
    # 创建所有协程任务
    tasks = [exec_id(id) for id in ids]
    # 并行执行所有任务
    results = await asyncio.gather(*tasks)
    print("所有MainTask状态:", results)

asyncio.run(main())

是否需要线程？ #

只有当load_tasks或其他方法包含CPU密集型操作时，才需要考虑用线程池（比如concurrent.futures.ThreadPoolExecutor）配合asyncio使用。如果只是Celery任务的提交和状态监控，纯asyncio足够高效。

内容的提问来源于stack exchange，提问作者AviC