在Python开发中，遇到需要执行数分钟甚至数小时的长时间计算任务时，单进程执行不仅效率有限，还会导致程序无响应，用户无法知晓计算进度。借助多进程技术可以将计算任务拆分到多个子进程执行，同时主进程负责实时接收并展示子进程返回的中间结果，兼顾计算效率和用户体验。
核心实现思路
整体方案分为三个部分：主进程负责创建子进程、接收结果并展示；子进程负责执行具体的长时间计算任务，并在计算过程中定期返回中间结果；进程间通过队列（Queue）实现安全的数据传递，避免资源竞争问题。
用到的核心模块
- multiprocessing：Python标准库中的多进程模块，提供进程创建、进程间通信的基础能力
- time：用于模拟计算耗时，实际场景中可替换为真实的计算逻辑
完整代码实现
子进程计算逻辑
子进程需要执行长时间的计算任务，并且在计算到一定阶段时，将中间结果放入进程队列中。以下示例中模拟了10轮计算，每轮计算耗时1秒，每完成一轮就返回一次当前进度和结果：
import time
from multiprocessing import Process, Queue
def long_time_calculate(queue):
# 模拟10轮长时间计算
total_round = 10
for round_num in range(1, total_round + 1):
# 模拟单轮计算耗时1秒
time.sleep(1)
# 计算本轮结果，这里模拟平方和计算
current_result = round_num * round_num
# 将进度和结果放入队列，传递给主进程
queue.put({
"round": round_num,
"total_round": total_round,
"current_result": current_result,
"progress": round_num / total_round * 100
})
# 计算完成后放入结束标识
queue.put(None)
主进程逻辑实现
主进程负责创建子进程和计算队列，启动子进程后持续从队列中读取结果并实时展示，直到收到子进程发送的结束标识：
def main():
# 创建进程间通信的队列
result_queue = Queue()
# 创建子进程，传入队列参数
calculate_process = Process(target=long_time_calculate, args=(result_queue,))
# 启动子进程
calculate_process.start()
print("长时间计算任务已启动，正在实时获取结果...")
while True:
# 从队列中获取结果
result = result_queue.get()
# 如果收到None，说明子进程计算完成
if result is None:
print("所有计算任务已完成")
break
# 实时展示当前计算结果
print(f"计算进度：{result['progress']:.1f}% | 当前轮次：{result['round']}/{result['total_round']} | 本轮结果：{result['current_result']}")
# 等待子进程结束
calculate_process.join()
if __name__ == "__main__":
main()
运行效果说明
执行上述代码后，会每秒输出一次当前的计算进度和结果，输出示例如下：
长时间计算任务已启动，正在实时获取结果...
计算进度：10.0% | 当前轮次：1/10 | 本轮结果：1
计算进度：20.0% | 当前轮次：2/10 | 本轮结果：4
计算进度：30.0% | 当前轮次：3/10 | 本轮结果：9
...
所有计算任务已完成
注意事项
- 队列的
get()方法是阻塞式的，当队列中没有数据时，主进程会暂停等待，直到子进程放入新数据，不会造成CPU空转 - 如果是Windows系统，多进程相关代码必须放在
if __name__ == "__main__":条件下执行，否则会出现进程无限递归创建的问题 - 如果计算任务产生的结果数据量较大，建议只传递必要的进度和摘要信息，避免队列占用过多内存
- 若需要终止计算任务，可以通过
terminate()方法停止子进程，但需要注意进程终止后队列中可能残留未处理的数据