Python多线程/multiprocessing无限循环任务调度实现求助

先帮你理清核心需求和之前方案的问题，再给出可运行的实现示例：

核心需求回顾 #

• 无限循环执行4个任务，耗时分别为：
  • T1（绿色）≈0.4s，持续运行，每次完成后立即启动新的T1，保留最新结果
  • T4（红色）≈0.02s，每次循环都获取最新结果
  • T2（蓝色）、T3（黑色）≈0.6-0.8s，必须依赖最新的T1+T4结果启动，且要在独立进程中并行执行，完成后合并结果
• 主线程不能被阻塞，要持续调度任务

之前方案的问题分析 #

1. ThreadPoolExecutor方案：Python的GIL（全局解释器锁）导致CPU密集型任务无法真正并行，T2和T3会串行执行，所以运行缓慢。如果你的任务是IO密集型，可能勉强能用，但你明确要求独立进程并行，所以这个方案不适用。

2. multiprocessing手动管理进程方案：

   • 使用process.join(X)会阻塞主线程，破坏了无限循环的调度逻辑
   • is_alive()状态判断不准确，因为join(X)只是等待X秒，进程可能还在运行但主线程提前退出等待
   • 队列的使用逻辑混乱，没有正确分离结果传递和状态判断

推荐实现方案：ProcessPoolExecutor + 状态跟踪 #

用concurrent.futures.ProcessPoolExecutor来管理进程，它封装了进程的创建和状态管理，比手动用multiprocessing.Process更简洁可靠。下面是符合你需求的完整代码：

import time
from concurrent.futures import ProcessPoolExecutor

# 模拟各个任务的实现
def worker1():
    time.sleep(0.4)  # 模拟T1耗时0.4s
    return f"T1_result_{int(time.time())}"  # 带时间戳标识最新结果

def worker2(input_data):
    time.sleep(0.7)  # 模拟T2耗时0.6-0.8s，取中间值0.7s
    return f"T2_output_{input_data}"

def worker3(input_data):
    time.sleep(0.7)  # 模拟T3耗时0.6-0.8s
    return f"T3_output_{input_data}"

def worker4():
    time.sleep(0.02)  # 模拟T4耗时0.02s
    return f"T4_result_{int(time.time())}"

def main():
    # 初始化进程池，最大进程数设为4（刚好容纳所有任务）
    with ProcessPoolExecutor(max_workers=4) as executor:
        # 跟踪各个任务的future对象和结果
        future_t1 = None
        result1 = None
        future_t2 = None
        future_t3 = None
        result4 = None

        while True:
            # 1. 获取最新的T4结果
            result4 = worker4()

            # 2. 管理T1的生命周期：持续运行，每次完成后立即启动新的T1
            if future_t1 is None or future_t1.done():
                if future_t1 is not None:
                    # 获取T1的最新结果
                    result1 = future_t1.result()
                    print(f"[更新] 拿到最新T1结果: {result1}")
                # 启动新的T1
                future_t1 = executor.submit(worker1)
                print(f"[启动] 新的T1任务已提交")

            # 3. 检查是否满足启动T2、T3的条件：有可用的result1+result4，且T2/T3未在运行
            if (result1 is not None and result4 is not None) and (future_t2 is None and future_t3 is None):
                # 并行启动T2和T3，传入最新的result1和result4
                future_t2 = executor.submit(worker2, f"{result1}_{result4}")
                future_t3 = executor.submit(worker3, f"{result1}_{result4}")
                print(f"[启动] T2和T3并行启动，输入: {result1}_{result4}")

            # 4. 检查T2、T3是否完成，合并结果
            if future_t2 is not None and future_t3 is not None:
                if future_t2.done() and future_t3.done():
                    result_t2 = future_t2.result()
                    result_t3 = future_t3.result()
                    end_result = f"{result_t2} + {result_t3}"
                    print(f"[完成] 合并T2和T3结果: {end_result}")
                    # 重置T2、T3的future对象，等待下一次启动
                    future_t2 = None
                    future_t3 = None

            # 主线程短暂休眠，避免空转占用过多CPU
            time.sleep(0.01)

if __name__ == "__main__":
    main()

关键细节解释 #

1. 进程池的选择：用ProcessPoolExecutor而不是线程池，因为它会创建独立的操作系统进程，避开GIL的限制，让T2和T3真正并行执行。
2. T1的持续运行：每次T1完成后立即启动新的实例，保证我们总能拿到最新的T1结果，符合你的需求。
3. 非阻塞状态检查：用done()方法判断任务是否完成，而不是join()，这样主线程不会被阻塞，可以持续处理其他调度逻辑。
4. 状态变量管理：通过跟踪future_t1、future_t2、future_t3这几个future对象，以及result1、result4结果变量，清晰控制任务的启动和结果的传递。
5. 避免CPU空转：主线程每次循环休眠0.01s，减少空转对CPU的占用。

这个方案完全符合你的伪代码逻辑，并且解决了之前方案的问题，能实现T2和T3的真正并行，同时保证任务依赖关系的正确性。

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