Python ThreadPoolExecutor如何实现类似Java FutureTask的依赖关系？

嘿，作为从Java转Python的开发者，我太懂这种要把熟悉的并发逻辑适配到新语言的感觉了！你想要的就是让combine函数必须等两个Future都返回结果后再执行对吧？结合你已经研究过的as_completed和wait，我给你整理了几种实用的实现方式，还有贴近Java习惯的封装思路，帮你快速上手~

这种方式最直观，和Java里等待多个Future的逻辑几乎一致，适合简单的依赖场景：

from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED

def task1():
    return "Result from Task 1"

def task2():
    return "Result from Task 2"

def combine(result1, result2):
    return f"Combined: {result1} + {result2}"

if __name__ == "__main__":
    with ThreadPoolExecutor(max_workers=2) as executor:
        # 提交两个任务到线程池
        future1 = executor.submit(task1)
        future2 = executor.submit(task2)
        
        # 等待两个Future全部完成，参数ALL_COMPLETED表示必须等所有都结束
        wait([future1, future2], return_when=ALL_COMPLETED)
        
        # 此时可以安全获取两个任务的结果，传给combine
        combined_result = combine(future1.result(), future2.result())
        print(combined_result)

这里的wait会阻塞到指定的Future都完成，之后你就可以放心调用result()获取结果，不用担心还没执行完的情况。

如果你的场景需要跟踪每个Future的完成状态（比如完成一个就打个日志），但最终还是要等所有结果齐了再执行combine，可以用as_completed：

from concurrent.futures import ThreadPoolExecutor, as_completed

def task1():
    return "Result from Task 1"

def task2():
    return "Result from Task 2"

def combine(result1, result2):
    return f"Combined: {result1} + {result2}"

if __name__ == "__main__":
    with ThreadPoolExecutor(max_workers=2) as executor:
        future1 = executor.submit(task1)
        future2 = executor.submit(task2)
        futures = [future1, future2]
        
        results = {}
        # 遍历已完成的Future，实时收集结果
        for future in as_completed(futures):
            if future == future1:
                results["task1"] = future.result()
                print("Task 1 finished!")
            else:
                results["task2"] = future.result()
                print("Task 2 finished!")
        
        # 所有结果收集完毕后，执行combine
        combined_result = combine(results["task1"], results["task2"])
        print(combined_result)

这种方式的好处是可以在每个任务完成时做一些中间操作，最后等结果都齐了再合并，适合复杂一点的业务场景。

如果你想更贴近Java里FutureTask处理依赖的友好写法，可以封装一个工具函数，让combine自动依赖指定的Future，不用手动写等待逻辑：

from concurrent.futures import ThreadPoolExecutor, Future
import threading

def task1():
    return "Result from Task 1"

def task2():
    return "Result from Task 2"

def combine(result1, result2):
    return f"Combined: {result1} + {result2}"

def when_all_done(futures, callback):
    """
    模拟Java风格的多Future依赖：等所有Future完成后执行回调
    :param futures: 需要等待的Future列表
    :param callback: 接收所有Future结果的回调函数
    """
    def wait_and_execute():
        try:
            # 等待所有Future完成，这里用遍历result()的方式，也可以替换成wait
            for future in futures:
                future.result()
            # 按顺序收集结果，传给回调
            results = [f.result() for f in futures]
            callback(*results)
        except Exception as e:
            # 可以在这里统一处理任务执行中的异常
            print(f"Task error: {e}")
    
    # 启动新线程等待，避免阻塞主线程（根据你的场景选择是否需要）
    threading.Thread(target=wait_and_execute).start()

if __name__ == "__main__":
    with ThreadPoolExecutor(max_workers=2) as executor:
        future1 = executor.submit(task1)
        future2 = executor.submit(task2)
        
        # 注册回调，等两个任务都完成后自动执行combine并打印结果
        when_all_done([future1, future2], lambda r1, r2: print(combine(r1, r2)))
        
        # 主线程可以继续处理其他业务
        print("Main thread is working on other tasks...")

这个封装函数可以帮你把等待逻辑和业务逻辑解耦，和Java里处理多个Future依赖的体验很像，还能扩展支持更多Future或者异常处理。

小提醒 #

不管用哪种方式，都别忘了处理异常：调用future.result()时会抛出任务执行过程中出现的异常，记得用try-except包裹，避免程序崩溃。如果combine本身也是耗时操作，也可以把它提交到线程池，不要阻塞主线程哦~

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