Python：如何复用长耗时函数旧结果，实现每5秒输出计算结果

针对Calculate_a耗时过长导致Sum无法按时输出的问题，以下是三种可行方案，根据你的任务类型选择：

1. 多线程方案（适合IO密集型任务） #

把Calculate_a放到后台线程执行，主线程每隔5秒调用Sum输出临时结果，直到a计算完成后输出最终值。

import threading
import time

def Calculate_a(x, y, z, t, result_container):
    # 替换为你的实际计算逻辑，这里用sleep模拟5小时耗时
    time.sleep(5)
    result_container['a'] = x + y + z + t  # 示例计算结果

def Sum(a, b):
    return a + b

if __name__ == "__main__":
    result = {'a': None}  # 用可变对象共享线程间结果
    x, y, z, t = 1, 2, 3, 4
    b = 10

    # 启动后台线程计算a
    calc_thread = threading.Thread(target=Calculate_a, args=(x, y, z, t, result))
    calc_thread.start()

    # 主线程周期性输出结果
    while True:
        current_a = result['a'] if result['a'] is not None else 0
        print(f"当前结果：{Sum(current_a, b)}")
        if result['a'] is not None:
            print(f"最终结果：{Sum(current_a, b)}")
            break
        time.sleep(5)
    calc_thread.join()

2. 异步IO方案（适合IO密集型任务，效率更高） #

如果Calculate_a是等待外部资源（比如网络请求、文件读写），用asyncio可以避免线程切换开销，更高效。

import asyncio

async def Calculate_a(x, y, z, t):
    # 替换为你的实际IO密集型逻辑，这里用sleep模拟耗时
    await asyncio.sleep(5)
    return x + y + z + t

def Sum(a, b):
    return a + b

async def periodic_sum(b, task_a):
    # 每隔5秒输出一次临时结果
    while not task_a.done():
        print(f"当前结果：{Sum(0, b)}")
        await asyncio.sleep(5)
    # 计算完成后输出最终结果
    final_a = await task_a
    print(f"最终结果：{Sum(final_a, b)}")

async def main():
    x, y, z, t = 1, 2, 3, 4
    b = 10
    task_a = asyncio.create_task(Calculate_a(x, y, z, t))
    await asyncio.gather(task_a, periodic_sum(b, task_a))

if __name__ == "__main__":
    asyncio.run(main())

3. 多进程方案（适合CPU密集型任务） #

如果Calculate_a是大量数学运算这类CPU密集型任务，Python的GIL会限制多线程效率，此时用多进程更合适，能利用多核CPU加速计算。

import multiprocessing
import time

def Calculate_a(x, y, z, t, result_queue):
    # 替换为你的实际CPU密集型逻辑，这里用sleep模拟耗时
    time.sleep(5)
    result_queue.put(x + y + z + t)

def Sum(a, b):
    return a + b

if __name__ == "__main__":
    x, y, z, t = 1, 2, 3, 4
    b = 10
    result_queue = multiprocessing.Queue()

    # 启动子进程计算a
    calc_process = multiprocessing.Process(target=Calculate_a, args=(x, y, z, t, result_queue))
    calc_process.start()

    # 主线程周期性输出结果
    while True:
        if not result_queue.empty():
            final_a = result_queue.get()
            print(f"最终结果：{Sum(final_a, b)}")
            break
        print(f"当前结果：{Sum(0, b)}")
        time.sleep(5)
    calc_process.join()

方案选择建议 #

• IO密集型任务（等待外部资源）：优先选asyncio，其次threading
• CPU密集型任务（大量运算）：选multiprocessing

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