Python Barrier 栅栏详解：基于最新版本的并发同步实践
- 一、Python Barrier 栅栏详解
- 1、 引言
- 2、 Barrier 是什么
- 2.1、 核心概念
- 3、 基本用法
- 3.1 、创建 Barrier
- 3.2、 线程等待
- 4、 Barrier 的完整 API
- 4.1 、构造参数
- 4.2、 实例方法
- 4.3 、实例属性
- 5、 进阶用法
- 5.1 、使用 action 回调
- 5.2、 获取到达顺序
- 5.3、 处理 BrokenBarrierError
- 6、 实战案例：分阶段并行计算
- 7、Barrier 与其它同步原语的对比
- 8、注意事项与最佳实践
- 8.1 、常见陷阱
- 8.2 、最佳实践
- 9、总结
- 二、代码示例
一、Python Barrier 栅栏详解
1、 引言
在多线程编程中，线程之间的同步与协作是保证程序正确性的关键。Python 的threading模块提供了多种同步原语，如Lock、Event、Semaphore等，而Barrier（栅栏）则是其中一种非常实用却常被忽视的同步机制。
Barrier的核心思想是：让一组线程相互等待，直到所有线程都到达某个“集合点”后，再一起继续执行。这种机制非常适合“分阶段并行计算”“多线程任务对齐”等场景。
本文基于 Python 最新稳定版本（3.12+），从原理、API、实战到注意事项，全面讲解Barrier的用法。
2、 Barrier 是什么
Barrier直译为“栅栏”或“屏障”，它会在代码中设置一道“栅栏”。当多个线程执行到这道栅栏时，必须全部到齐才能放行，否则先到的线程会阻塞等待。
与Event不同，Barrier是一次性的同步点，且具有“可重用”特性（通过reset()或自动重置）。与Semaphore不同，Barrier强调的是所有参与者都到达，而不是“允许 N 个线程进入”。
2.1、 核心概念
- parties：参与栅栏的线程总数。
- wait()：线程调用此方法表示“我已到达栅栏”，并阻塞直到所有 parties 都到达。
- broken：当某个线程在等待过程中被取消或超时，栅栏会进入 broken 状态，所有等待线程会收到
BrokenBarrierError。
3、 基本用法
3.1 、创建 Barrier
importthreading# 创建一个需要 3 个线程参与的栅栏barrier=threading.Barrier(3)3.2、 线程等待
importthreadingimporttime barrier=threading.Barrier(3)defworker(name):print(f"{name}开始工作")time.sleep(1)# 模拟耗时操作print(f"{name}到达栅栏")barrier.wait()# 等待其他线程print(f"{name}通过栅栏，继续执行")threads=[threading.Thread(target=worker,args=(f"线程-{i}",))foriinrange(3)]fortinthreads:t.start()fortinthreads:t.join()运行结果：
线程-0 开始工作 线程-1 开始工作 线程-2 开始工作 线程-0 到达栅栏 线程-1 到达栅栏 线程-2 到达栅栏 线程-2 通过栅栏，继续执行 线程-0 通过栅栏，继续执行 线程-1 通过栅栏，继续执行可以看到，三个线程都到达栅栏后，才一起继续执行。
4、 Barrier 的完整 API
4.1 、构造参数
threading.Barrier(parties,action=None,timeout=None)| 参数 | 说明 |
|---|---|
parties | 参与线程数量，必须大于 0 |
action | 当所有线程到达时，由最后一个到达的线程执行的回调函数 |
timeout | 每个wait()的默认超时时间（秒） |
4.2、 实例方法
| 方法 | 说明 |
|---|---|
wait(timeout=None) | 阻塞直到所有线程到达；返回一个整数，表示当前线程是第几个到达的（0 表示第一个） |
reset() | 将栅栏重置为初始状态，所有等待线程会收到BrokenBarrierError |
abort() | 将栅栏置为 broken 状态，所有等待线程会收到BrokenBarrierError |
4.3 、实例属性
| 属性 | 说明 |
|---|---|
parties | 参与线程数量 |
n_waiting | 当前正在等待的线程数量 |
broken | 栅栏是否处于 broken 状态 |
5、 进阶用法
5.1 、使用 action 回调
action参数允许在所有线程到达时执行一次特定操作，非常适合“阶段汇总”场景：
importthreadingimporttimedefmerge_results():print("=== 所有线程已到达，执行汇总操作 ===")barrier=threading.Barrier(3,action=merge_results)defworker(name):time.sleep(1)barrier.wait()print(f"{name}继续执行")threads=[threading.Thread(target=worker,args=(f"线程-{i}",))foriinrange(3)]fortinthreads:t.start()fortinthreads:t.join()运行结果：
=== 所有线程已到达，执行汇总操作 === 线程-2 继续执行 线程-0 继续执行 线程-1 继续执行注意：action只执行一次，且由最后一个到达的线程执行。
5.2、 获取到达顺序
wait()的返回值表示当前线程是第几个到达的：
importthreadingimporttime barrier=threading.Barrier(3)defworker(name):time.sleep(1)index=barrier.wait()print(f"{name}是第{index}个到达的")threads=[threading.Thread(target=worker,args=(f"线程-{i}",))foriinrange(3)]fortinthreads:t.start()fortinthreads:t.join()5.3、 处理 BrokenBarrierError
当某个线程在等待时被取消或超时，栅栏会进入 broken 状态：
importthreadingimporttime barrier=threading.Barrier(3)defworker(name):try:barrier.wait(timeout=2)print(f"{name}通过栅栏")exceptthreading.BrokenBarrierError:print(f"{name}检测到栅栏已损坏")# 只启动 2 个线程，第 3 个永远不会到达t1=threading.Thread(target=worker,args=("线程-1",))t2=threading.Thread(target=worker,args=("线程-2",))t1.start()t2.start()t1.join()t2.join()运行结果：
线程-1 检测到栅栏已损坏 线程-2 检测到栅栏已损坏6、 实战案例：分阶段并行计算
下面通过一个“分阶段数据处理”的案例，展示Barrier的实际应用价值：
importthreadingimporttimeimportrandomclassPipeline:def__init__(self,num_workers):self.num_workers=num_workers self.stage1_results={}self.stage2_results={}# 两个阶段的栅栏self.barrier1=threading.Barrier(num_workers)self.barrier2=threading.Barrier(num_workers)defstage1(self,worker_id):"""第一阶段：数据采集"""time.sleep(random.uniform(0.5,1.5))data=random.randint(1,100)self.stage1_results[worker_id]=dataprint(f"Worker-{worker_id}完成数据采集:{data}")self.barrier1.wait()# 等待所有 worker 完成采集returndatadefstage2(self,worker_id,data):"""第二阶段：数据处理"""# 模拟依赖其他 worker 的数据total=sum(self.stage1_results.values())result=data*total self.stage2_results[worker_id]=resultprint(f"Worker-{worker_id}完成数据处理:{result}")self.barrier2.wait()# 等待所有 worker 完成处理returnresultdefrun(self,worker_id):data=self.stage1(worker_id)self.stage2(worker_id,data)pipeline=Pipeline(3)threads=[threading.Thread(target=pipeline.run,args=(i,))foriinrange(3)]fortinthreads:t.start()fortinthreads:t.join()print(f"最终结果:{pipeline.stage2_results}")这个案例展示了Barrier在多阶段流水线中的典型用法：每个阶段结束后，所有线程必须对齐，才能进入下一阶段。
7、Barrier 与其它同步原语的对比
| 特性 | Barrier | Event | Semaphore | Lock |
|---|---|---|---|---|
| 同步方式 | 所有线程互相等待 | 一个线程通知其他线程 | 控制并发访问数量 | 互斥访问共享资源 |
| 参与者数量 | 固定（parties） | 不固定 | 不固定 | 不固定 |
| 可重用 | 支持（reset） | 支持（clear/set） | 支持 | 支持 |
| 典型场景 | 分阶段并行计算 | 事件通知 | 连接池限流 | 临界区保护 |
8、注意事项与最佳实践
8.1 、常见陷阱
线程数量不匹配：启动的线程数少于
parties，会导致所有线程永久阻塞。务必确保线程数量与parties一致。异常导致栅栏损坏：如果某个线程在
wait()前抛出异常，其他线程会一直等待。建议使用try/finally或abort()兜底。超时设置：生产环境建议始终设置
timeout，避免因某个线程卡死导致整个程序挂起。
8.2 、最佳实践
- 使用
with语句或try/finally确保栅栏状态正确。 - 在
action回调中执行汇总操作时，注意线程安全。 - 对于可重用的多阶段任务，考虑封装
Barrier到类中统一管理。
importthreading barrier=threading.Barrier(3,timeout=5)defsafe_worker(name):try:barrier.wait()exceptthreading.BrokenBarrierError:print(f"{name}: 栅栏已损坏，任务终止")exceptExceptionase:print(f"{name}: 发生异常{e}")barrier.abort()# 通知其他线程9、总结
Barrier是 Python 多线程编程中强大的同步工具，特别适合需要多线程对齐的场景。通过本文的讲解，你应该已经掌握了：
Barrier的核心原理与基本用法- 完整的 API 与参数说明
action回调、超时处理等进阶技巧- 分阶段并行计算的实战案例
- 与其他同步原语的对比与选型建议
在实际项目中，合理使用Barrier可以显著简化多线程协作逻辑，提升代码的可读性与健壮性。希望本文能帮助你在并发编程的道路上更进一步。
二、代码示例
importthreadingimporttime# 栅栏集齐后执行的回调函数defbarrier_callback():print("\n👉【栅栏回调函数执行】所有线程全部准备就绪！\n")defworker(barrier:threading.Barrier,tid:int):print(f"线程[{tid}]：正在执行前置初始化")time.sleep(0.2*tid)print(f"线程[{tid}]：到达栅栏，当前等待线程数 ={barrier.n_waiting}")try:# wait 返回序列号：0 ~ parties‑1serial=barrier.wait(timeout=5)exceptthreading.BrokenBarrierError:print(f"线程[{tid}]：栅栏损坏 / 等待超时，退出任务")returnprint(f"线程[{tid}]：冲破栅栏，序列号={serial}，开始并行执行业务\n")defdemo_normal_complete():print("="*70)print("【场景1：正常凑齐线程，同步放行 + action回调】")bar=threading.Barrier(parties=3,action=barrier_callback)thread_list=[]foriinrange(3):t=threading.Thread(target=worker,args=(bar,i))thread_list.append(t)t.start()fortinthread_list:t.join()print("场景1执行完毕\n")defdemo_reset_reuse():print("="*70)print("【场景2：reset()重置栅栏，多轮重复使用】")bar=threading.Barrier(parties=2,action=barrier_callback)defone_round(round_id):print(f"\n------ 第{round_id}轮同步 ------")t1=threading.Thread(target=worker,args=(bar,10+round_id))t2=threading.Thread(target=worker,args=(bar,20+round_id))t1.start()t2.start()t1.join()t2.join()bar.reset()# 重置计数器，栅栏恢复初始状态one_round(1)one_round(2)print("场景2执行完毕\n")defdemo_abort_broken():print("="*70)print("【场景3：abort() 主动破坏栅栏，抛出 BrokenBarrierError】")bar=threading.Barrier(parties=2)defabort_task():try:bar.wait()exceptthreading.BrokenBarrierError:print("捕获异常：BrokenBarrierError，栅栏已被破坏")t=threading.Thread(target=abort_task)t.start()time.sleep(0.3)bar.abort()# 损坏栅栏t.join()# 损坏后的栅栏，直接wait依然报错，必须reset恢复try:bar.wait(timeout=1)exceptthreading.BrokenBarrierError:print("损坏状态下直接wait依然异常，执行reset恢复栅栏")bar.reset()print("场景3执行完毕\n")defdemo_wait_timeout():print("="*70)print("【场景4：线程数量不足，wait超时抛出异常】")bar=threading.Barrier(parties=2)deftimeout_task():try:bar.wait(timeout=1)exceptthreading.BrokenBarrierError:print("捕获异常：等待超时，栅栏损坏")t=threading.Thread(target=timeout_task)t.start()t.join()bar.reset()print("场景4执行完毕\n")if__name__=="__main__":demo_normal_complete()demo_reset_reuse()demo_abort_broken()demo_wait_timeout()D:\user\01417804\桌面\PythonProject\.venv\Scripts\python.exe D:\user\01417804\桌面\PythonProject\main.py======================================================================【场景1：正常凑齐线程，同步放行+action回调】 线程[0]：正在执行前置初始化 线程[0]：到达栅栏，当前等待线程数=0线程[1]：正在执行前置初始化 线程[2]：正在执行前置初始化 线程[1]：到达栅栏，当前等待线程数=1线程[2]：到达栅栏，当前等待线程数=2👉【栅栏回调函数执行】所有线程全部准备就绪！ 线程[2]：冲破栅栏，序列号=2，开始并行执行业务 线程[0]：冲破栅栏，序列号=0，开始并行执行业务 线程[1]：冲破栅栏，序列号=1，开始并行执行业务 场景1执行完毕======================================================================【场景2：reset()重置栅栏，多轮重复使用】------第1轮同步------线程[11]：正在执行前置初始化 线程[21]：正在执行前置初始化 线程[11]：到达栅栏，当前等待线程数=0线程[21]：到达栅栏，当前等待线程数=1👉【栅栏回调函数执行】所有线程全部准备就绪！ 线程[21]：冲破栅栏，序列号=1，开始并行执行业务 线程[11]：冲破栅栏，序列号=0，开始并行执行业务------第2轮同步------线程[12]：正在执行前置初始化 线程[22]：正在执行前置初始化 线程[12]：到达栅栏，当前等待线程数=0线程[22]：到达栅栏，当前等待线程数=1👉【栅栏回调函数执行】所有线程全部准备就绪！ 线程[22]：冲破栅栏，序列号=1，开始并行执行业务 线程[12]：冲破栅栏，序列号=0，开始并行执行业务 场景2执行完毕======================================================================【场景3：abort()主动破坏栅栏，抛出 BrokenBarrierError】 捕获异常：BrokenBarrierError，栅栏已被破坏 损坏状态下直接wait依然异常，执行reset恢复栅栏 场景3执行完毕======================================================================【场景4：线程数量不足，wait超时抛出异常】 捕获异常：等待超时，栅栏损坏 场景4执行完毕 进程已结束，退出代码为0
