Python的线程21 Barrier类再次解析

Posted 2022-11-18 雷学委

正式的Python专栏第58篇，同学站住，别错过这个从0开始的文章！

这篇继续介绍threading库里面的类，threading.Barrier类。

这个类很好用，直接指定数值，通过调用其wait方法，就轻松实现多个线程同步等待。

大家也看到学委之前用condition的实现（需要正确的使用notify和wait，还有加上wait(timeout) 来实现十个运动员等待）明显比Barrier类绕了一些。

继续看看Barrier类

再次阅读一下Barrier类的构造方法（如下），我们看到，它内部持有了Condition对象和_count, _parties。

基本上Barrier利用了这三个实现了一个等待释放的机制，下面看看wait源码。

另外，还有其他参数像action，这个会再整个Barrier被release之前由某一个线程调用。

def __init__(self, parties, action=None, timeout=None):
    self._cond = Condition(Lock())
    self._action = action
    self._timeout = timeout
    self._parties = parties
    self._state = 0 #0 filling, 1, draining, -1 resetting, -2 broken
    self._count = 0

学委准备了下面这个小demo方法，读者可以在前面分享的代码中贴上。

def post_action():
    print("运动员调用了barrier释放后的方法- %s" % threading.current_thread().name)
barrier = threading.Barrier(parties=10,action=post_action)

重新运行代码可以看到，只有某一个运动员线程（哪个运动员是随机的，看实际项目中是那个线程最后调用wait方法的）执行了post_aciton方法。

下面是某一次执行的效果展示：

重点是这个wait方法

它像Condition一样也支持timeout（也就是线程数量凑不够，在一个固定等待延迟后强制执行wait后面的代码）。

def wait(self, timeout=None):
   if timeout is None:
       timeout = self._timeout
   with self._cond:
       self._enter() # Block while the barrier drains.
       index = self._count
       self._count += 1
       try:
           if index + 1 == self._parties:
               # We release the barrier
               self._release()
           else:
               # We wait until someone releases us
               self._wait(timeout)
           return index
       finally:
           self._count -= 1
           # Wake up any threads waiting for barrier to drain.
           self._exit()

这段代码也非常直观，每一个线程调用一次wait方法，Barrier对象持有的_count自增1（从0开始）。

知道_count 为_parties-1的时候，Barrier调用_release释放内部持有的锁。

好，继续看一下_release 函数，这个更加简单了，重点我们看到：self._cond.notify_all()，这句执行了。

它调用了notify_all函数，释放了其他调用的condition.wait方法。

这就是Barrier类的内部实现。

def _release(self):
    try:
        if self._action:
            self._action()
        # enter draining state
        self._state = 1
        self._cond.notify_all()
    except:
        #an exception during the _action handler.  Break and reraise
        self._break()
        raise

另外，这里我们也看到_action方法被调用了，这也验证了学委上面贴上的demo
运行图，10号运动员打印了post_action方法。

补充一下with方式的使用Condition，也可以应用在Lock/Rlock对象。

像下面的代码：

with xuewei_lock:
    do_something()

等效于：

xuewei_lock.acquire()
do_something()
xuewei_lock.release()

这种风格让这个锁的使用看起来更加优雅，也不容易漏了。

毕竟有时候方法不小心写多了，后面容易就给忘记了，特别是多人协作的项目上。（后来者非常有可能在项目非常紧急的情况下，修改代码导致某些情况锁没有被释放，产生系统故障，这就非常可惜。）

总结

Barrier类内部使用了Condition这个资源控制单元来实现，加上一个阀值和计数器变量，实现多线程‘屏障’功能。

其他方法读者可以自行阅读，加深对这个类的理解。

喜欢Python的朋友，请关注学委的 Python基础专栏 or Python入门到精通大专栏

持续学习持续开发，我是雷学委！
编程很有趣，关键是把技术搞透彻讲明白。
欢迎关注微信，点赞支持收藏!