如何将子任务外包至第二CPU并按需获取计算结果？

针对你需要在主循环中异步执行非必要耗时子任务的需求，用Python的multiprocessing模块就能实现，核心思路是异步提交子任务、非阻塞检查任务状态、按需获取结果并提交新任务，以下是两种具体实现方式：

方式一：使用进程池（Pool）的异步提交 #

进程池适合重复执行同类子任务，自带进程管理，无需手动维护进程生命周期：

import multiprocessing
import time

# 定义耗时子任务
def long_running_task(task_id):
    print(f"启动子任务 {task_id}")
    time.sleep(3)  # 模拟耗时操作
    return f"子任务 {task_id} 结果"

if __name__ == "__main__":
    # 创建单进程池，专门分配给第二CPU处理子任务
    pool = multiprocessing.Pool(processes=1)
    # 提交第一个子任务
    current_task = pool.apply_async(long_running_task, args=(1,))
    task_id = 2

    while True:
        # 主循环执行核心任务
        print("主循环处理其他任务...")
        time.sleep(0.5)  # 模拟核心任务耗时

        # 非阻塞检查子任务是否完成
        if current_task.ready():
            try:
                # 获取子任务结果（此时不会阻塞，因为ready()已确认完成）
                result = current_task.get()
                print(f"拿到子任务结果: {result}")
                # 立即提交新的子任务
                current_task = pool.apply_async(long_running_task, args=(task_id,))
                task_id += 1
            except Exception as e:
                print(f"子任务执行出错: {str(e)}")
                # 出错后重新提交任务
                current_task = pool.apply_async(long_running_task, args=(task_id,))
                task_id += 1

关键说明 #

• apply_async()：异步提交任务，主进程不会等待子任务完成，直接继续执行
• ready()：非阻塞检查任务是否完成，不会拖慢主循环
• 进程池自动管理子进程，无需手动处理启动/关闭逻辑

方式二：使用Process + Queue #

如果需要更灵活的进程控制（比如自定义子进程行为），可以用Process结合队列传递结果：

import multiprocessing
import time
import queue

# 耗时子任务：执行完成后将结果放入队列
def long_running_task(task_id, result_queue):
    print(f"启动子任务 {task_id}")
    time.sleep(3)
    result_queue.put((task_id, f"子任务 {task_id} 结果"))

if __name__ == "__main__":
    result_queue = multiprocessing.Queue()
    task_id = 1
    # 启动第一个子进程
    sub_process = multiprocessing.Process(target=long_running_task, args=(task_id, result_queue))
    sub_process.start()
    task_id += 1

    while True:
        # 主循环执行核心任务
        print("主循环处理其他任务...")
        time.sleep(0.5)

        # 非阻塞检查队列是否有结果
        try:
            task_id_done, result = result_queue.get(block=False)
            print(f"拿到子任务 {task_id_done} 的结果: {result}")
            # 启动新的子进程
            sub_process = multiprocessing.Process(target=long_running_task, args=(task_id, result_queue))
            sub_process.start()
            task_id += 1
        except queue.Empty:
            # 队列空，说明子任务未完成，继续主循环
            pass

        # 可选：检查子进程是否意外退出，自动重启
        if not sub_process.is_alive():
            print(f"子进程意外终止，重启任务 {task_id}")
            sub_process = multiprocessing.Process(target=long_running_task, args=(task_id, result_queue))
            sub_process.start()
            task_id += 1

关键说明 #

• Queue：跨进程安全的结果传递容器，主进程通过非阻塞get()检查结果
• is_alive()：监控子进程状态，避免子任务异常退出后断档

两种方式都能实现你的需求：主循环持续执行核心任务，耗时子任务在独立CPU上运行，主进程按需获取结果并提交新任务，完全不会被子任务拖慢。

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