Python多线程中如何让线程等待其他线程完成指定任务？

嘿，这个场景我太熟悉了——多个线程各自完成计算任务后，需要集体同步才能进入后续的比较环节，而且不能用join？没问题，咱们来聊聊几个靠谱的方案，都是实际项目里用过的：

方案一：用threading.Barrier（最推荐！） #

Barrier就是专门为这种“所有线程必须到达某个 checkpoint 才能继续执行”的场景设计的，简直是为你的需求量身定做。你只需要初始化一个屏障，指定需要等待的线程数量，每个线程完成计算后调用wait()，它就会自动阻塞直到最后一个线程也完成计算，然后所有线程一起解锁进入下一阶段。

举个Python的例子：

import threading

# 共享结果列表（注意：列表的append操作在CPython里是线程安全的，要是修改元素记得加锁）
calculation_results = []
# 初始化屏障，参数是需要同步的线程总数
sync_barrier = threading.Barrier(3)  # 假设我们有3个工作线程

def distance_worker(thread_id):
    # 第一步：执行距离计算并存储结果
    print(f"线程{thread_id}正在计算距离...")
    simulated_distance = thread_id * 15  # 模拟计算结果
    calculation_results.append((thread_id, simulated_distance))
    
    # 等待所有线程完成计算
    print(f"线程{thread_id}计算完毕，等待其他线程同步...")
    sync_barrier.wait()
    
    # 第二步：和其他线程的结果比较并执行操作
    print(f"线程{thread_id}开始处理比较逻辑...")
    for tid, dist in calculation_results:
        if tid != thread_id:
            print(f"线程{thread_id}对比线程{tid}的距离结果：{dist}")

# 创建并启动线程
threads = []
for i in range(3):
    worker_thread = threading.Thread(target=distance_worker, args=(i,))
    threads.append(worker_thread)
    worker_thread.start()

# 主线程等待所有工作线程结束（这步是主线程的等待，不是线程间的同步）
for t in threads:
    t.join()

这个方案的优点是代码简洁、语义明确，不需要自己维护计数器或者状态，Python的标准库已经帮你处理了所有同步细节，几乎不会出错。

方案二：用Event + 共享计数器（手动控制同步） #

如果因为某些限制不能用Barrier（比如老版本Python环境），那可以用threading.Event配合一个受锁保护的计数器来实现。思路是：每个线程完成计算后就把计数器加1，当计数器达到线程总数时，触发事件通知所有线程可以开始比较。

示例代码：

import threading

calculation_results = []
# 用于通知所有线程计算完成的事件
all_calculated_event = threading.Event()
# 记录完成计算的线程数，用锁保护避免竞态条件
completed_threads = 0
count_lock = threading.Lock()
total_thread_count = 3

def distance_worker(thread_id):
    global completed_threads
    
    # 计算阶段
    print(f"线程{thread_id}正在执行距离计算...")
    simulated_distance = thread_id * 15
    calculation_results.append((thread_id, simulated_distance))
    
    # 更新完成计数器，注意加锁
    with count_lock:
        completed_threads += 1
        # 当所有线程都完成时，触发事件
        if completed_threads == total_thread_count:
            all_calculated_event.set()
    
    # 等待事件触发（也就是等待所有线程完成计算）
    print(f"线程{thread_id}等待其他线程完成计算...")
    all_calculated_event.wait()
    
    # 比较阶段
    print(f"线程{thread_id}开始处理比较逻辑...")
    for tid, dist in calculation_results:
        if tid != thread_id:
            print(f"线程{thread_id} vs 线程{tid}：距离={dist}")

# 启动线程
threads = []
for i in range(total_thread_count):
    worker_thread = threading.Thread(target=distance_worker, args=(i,))
    threads.append(worker_thread)
    worker_thread.start()

# 主线程等待所有线程结束
for t in threads:
    t.join()

这个方案更灵活，你可以根据需求调整触发条件（比如不需要等所有线程，只需要等N个），但需要自己处理锁和计数器，相对容易出错，所以写的时候要注意锁的范围。

方案三：用队列收集结果（备选方案） #

你也可以用queue.Queue来收集所有线程的计算结果，每个线程完成计算后把结果放入队列，然后每个线程等待队列中的结果数量达到线程总数后再开始比较。不过要注意：Queue.qsize()在多线程环境下不是100%可靠，所以最好结合锁来辅助判断，或者用队列的join()方法配合任务标记，但这个方法不如前两个直接，所以只作为备选。

内容的提问来源于stack exchange，提问作者M.SEF