Python threading.Barrier 栅栏对象

说明

《Python 教程》持续更新中，提供建议、纠错、催更等加作者微信: gairuo123（备注：pandas教程）和关注公众号「盖若」ID: gairuo。跟作者学习，请进入 Python学习课程。欢迎关注作者出版的书籍：《深入浅出Pandas》和《Python之光》。

threading.Barrier 是一种同步原语，允许指定数量的线程在栅栏处等待，直到所有线程都达到栅栏。只有当所有线程都到达时，才会继续执行。在 Python 的 threading 模块中，Barrier（栅栏对象）用于实现多个线程在某个点进行同步，直到达到指定数量的线程都到达该点后，所有线程才可以继续执行。

栅栏对象

构造方法:

Barrier(parties, action=None, timeout=None)

创建一个栅栏对象。

参数:

parties: 需要同步的线程数量。
action: 所有线程都到达栅栏时执行的可选回调函数。
timeout: 等待线程超时的可选时间。

常用方法：

wait(timeout=None): 使线程在栅栏处等待，直到指定数量的线程到达。
reset(): 重置栅栏为初始状态。
abort(): 中止栅栏，所有等待线程将收到 BrokenBarrierError。
parties: 返回栅栏需要的线程总数。
n_waiting: 返回当前在栅栏处等待的线程数量。

使用场景：

线程同步: 在并行计算中，确保多个线程在某个阶段都已完成，然后一起继续执行。
阶段性任务: 用于将任务分为多个阶段，每个阶段开始前等待所有线程完成上一阶段。

示例代码

示例如下：

import threading
import time

def stage_task(barrier, n):
    print(f"线程 {n} 开始阶段 1")
    time.sleep(1)  # 模拟阶段 1 的工作
    print(f"线程 {n} 完成阶段 1")
    barrier.wait()  # 阶段 1 完成，等待其他线程

    print(f"线程 {n} 开始阶段 2")
    time.sleep(1)  # 模拟阶段 2 的工作
    print(f"线程 {n} 完成阶段 2")

barrier = threading.Barrier(3)  # 需要3个线程同步

threads = []
for i in range(3):
    t = threading.Thread(target=stage_task, args=(barrier, i))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

再如：

import threading
import time

barrier = threading.Barrier(3)  # 三个线程同步

def thread_function(name):
    print(f"{name} 正在执行任务")
    time.sleep(2)
    print(f"{name} 到达栅栏，等待其他线程")
    barrier.wait()
    print(f"{name} 继续执行任务")

threads = [
    threading.Thread(target=thread_function, args=("线程 1",)),
    threading.Thread(target=thread_function, args=("线程 2",)),
    threading.Thread(target=thread_function, args=("线程 3",))
]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

综合案例

假设我们有一个数据处理的任务，多个线程分别处理一部分数据，处理完成后需要进行数据的汇总和分析，使用栅栏来确保所有数据处理线程都完成后再进行汇总操作。

import threading
import time

barrier = threading.Barrier(3)  # 假设有三个数据处理线程

def data_processing(name):
    print(f"{name} 开始处理数据")
    time.sleep(2)
    print(f"{name} 处理数据完成，到达栅栏")
    barrier.wait()
    if name == "线程 1":
        # 模拟汇总操作
        print("开始汇总数据")

threads = [
    threading.Thread(target=data_processing, args=("线程 1",)),
    threading.Thread(target=data_processing, args=("线程 2",)),
    threading.Thread(target=data_processing, args=("线程 3",))
]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

在上述案例中，三个线程完成数据处理后通过栅栏进行同步，然后由指定的线程（这里是线程 1）进行数据汇总操作。

参考

https://docs.python.org/zh-cn/3/library/threading.html#barrier-objects

< Timer 定时器对象 Python 模块 threading 基于线程的并行

更新时间：June 30, 2024, 6:19 p.m. 标签：python 线程 threading 栅栏对象