go同步原语Phaser和Barrier区别

Phaser 同步原语

Java 中的 Phaser 是一个同步原语，它可以让多个线程在某个时刻同步执行。它和 Barrier 有点类似，但是它比 Barrier 更加灵活。

举一个例子哈，比如足球迷特别喜欢的欧洲冠军联赛，它的赛制就分为多个阶段：

欧洲冠军联赛由欧冠资格赛、欧冠附加赛和欧冠正赛三部分组成。 欧冠资格赛，分为预赛轮（preliminary round）、第一轮资格赛（first qualifying round）、第二轮资格赛（second qualifying round）和第三轮资格赛（third qualifying round）。第三轮资格赛的优胜的 10 支球队进入欧冠附加赛，附加赛优胜的 6 支球队（冠军之路 4 队、联赛之路 2 队）将和 26 支自动晋级的队伍一起，参加欧冠小组赛。 欧冠正赛分为小组赛、1/8 决赛、1/4 决赛、半决赛和决赛。

所以欧冠联赛分成了多个阶段，每一个阶段，会有一些球队参加，等到下一阶段，淘汰了一部分球队，又会有新的球队加入，每个阶段的球队都会有变化。这种情况非常适合使用 Phaser 来模拟。

Phaser 和 CyclicBarrier对比

Phaser 和 CyclicBarrier的功能非常的相似，都是应用于多个参与者多阶段处理问题的场景，每个阶段都有障碍点，在障碍点需要等待所有的参与者到齐后才能进入下一个阶段。但是Phaser比CyclicBarrier更加灵活，CyclicBarrier的参与者数量是固定的，所以初始化CyclicBarrier的时候就需要设定参与者的数量，而Phaser的参与者数量是可以动态变化的，每个阶段完成后参与者可以选择离开，新的参与者也可以加入进来，所以上面欧冠的例子非常使用Phaser来模拟。

一旦一个同步原语的功能不是那么通用，而是面向非常细分的场景，那么它的使用范围非常有限，因为大部分场景我们都会使用WaitGroup、channel甚至CyclicBarrier去解决，但是针对参与者需要动态变化的场景，我们使用Phaser如鱼得水，比自己再封装和实现类似Phaser的功能更方便。正所谓“技多不压身”，我们多了解一些同步原语，在解决问题的时候就会更加得心应手。

Go 标准库和扩展库中都没有实现，第三方库也鲜有实现，但是 Java 中有，我们可以参考 Java 中的实现，自己实现了一个，比如 github.com/smallnest/exp/sync/Phaser[1]，当然针对 Java 复杂的实现做了精简，不再支持 Phaser 的父子关系，函数名也做了简化，将Register/Deregister改为Join/Leave等，如果你之前不了解 Java 的 Phaser，可以看看Java Phaser[2]。

方法

我们看看它的方法：

func NewPhaser(parties int32) *Phaser
func NewPhaserWithAction(parties int32, arriveAction func(parties int32) error) *Phaser
func (p *Phaser) Arrive() int32
func (p *Phaser) ArriveAndLeave() int32
func (p *Phaser) Wait(phase int) int32
func (p *Phaser) ArriveAndWait() int32
func (p *Phaser) ForceTermination()
func (p *Phaser) IsTerminated() bool
func (p *Phaser) Join() int32
func (p *Phaser) BulkJoin(parties int32) int32
func (p *Phaser) Leave() int32
func (p *Phaser) Arrived() int32
func (p *Phaser) Parties() int32
func (p *Phaser) Phase() int32

分成几个部分来看

• 初始化

  • NewPhaser：初始化一个 Phaser，指定参与者的数量。

  • NewPhaserWithAction：初始化一个 Phaser，指定参与者的数量，以及每个阶段的障碍点到达后的回调函数。

• 动作

  • Arrive：参与者到达障碍点，但是不等待其他参与者，直接返回当前阶段。阶段的编号从 0 开始，每进入一个新的阶段，阶段编号会自增 1。

  • ArriveAndLeave：参与者到达障碍点，但是不等待其他参与者，直接返回当前阶段，并且离开 Phaser。

  • Wait：等待指定的阶段，如果指定的阶段已经完成，直接返回，否则等待指定的阶段完成后返回。

  • ArriveAndWait：参与者到达障碍点，等待其他参与者到达障碍点，然后返回当前阶段。

• 加入和离开

  • Join：参与者加入 Phaser，参与者的数量会自增 1。

  • Leave：参与者离开 Phaser，参与者的数量会自减 1。

  • BulkJoin：批量加入参与者，参与者的数量会自增指定的数量。

• 终止

  • ForceTermination：强制终止 Phaser，所有的参与者都会离开 Phaser。

• 查询

  • Arrived：返回当前阶段已经到达障碍点的参与者数量。

  • Parties：返回当前 Phaser 中参与者的数量。

  • Phase：返回当前阶段的编号。

  • IsTerminated：返回 Phaser 是否已经终止。

解释代码逻辑

举一个例子,我在代码中加上注释，来解释代码的逻辑：

func TestPhaser_phase(t *testing.T) {
 // 模拟一个有三个阶段的场景，其中一个参与者(id=1)在参加完第一阶段就离开了
    // 初始化一个Phaser，指定参与者的数量为0
 phaser := NewPhaser(0)
    // 在整个过程中有10个参与者
 var wg sync.WaitGroup
 wg.Add(10)
 for i := 0; i < 10; i++ {
  phaser.Join()
  go func(id int) {
   defer wg.Done()
   t.Logf("goroutine %d started\n", id)
   // phase == 0， 第一阶段，等待其他参与者到达
   phase := phaser.ArriveAndWait()
   assert.Equal(t, int32(1), phase) // 第一阶段完成，返回下一阶段的编号1
   t.Logf("goroutine %d finished phase 0， enter phase 1\n", id)
   if id == 1 { // #1 goroutine退出后面的阶段
    phaser.Leave()
    t.Logf("goroutine %d exit after phase 0\n", id)
    return
   }
   time.Sleep(time.Duration(rand.Intn(10) * int(time.Second)))
   // phase == 1， 第二阶段，等待其他参与者到达
   assert.Equal(t, int32(1), phaser.Phase())
   phase = phaser.ArriveAndWait()
   assert.Equal(t, int32(2), phase) // 第二阶段完成，返回下一阶段的编号2
   t.Logf("goroutine %d finished phase 1， enter phase 2\n", id)
   time.Sleep(time.Duration(rand.Intn(10) * int(time.Second)))
            // phase == 2， 第三阶段，等待其他参与者到达
   phase = phaser.ArriveAndWait()
   assert.Equal(t, int32(3), phase) // 第三阶段完成，返回下一阶段的编号3
   t.Logf("goroutine %d finished phase 2， enter phase 3\n", id)
  }(i)
 }
 wg.Wait() // 所有的参与者都完成了各阶段。 9个参与了三个阶段，一个参与了第一阶段
    phaser.ForceTermination() // 强制终止Phaser
 t.Logf("phaser terminated")
}

参考资料

[1]

github.com/smallnest/exp/sync/Phaser: https://github.com/smallnest/exp/blob/master/sync/phaser.go

[2]

Java Phaser: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Phaser.html

以上就是go同步原语Phaser和Barrier区别的详细内容，更多关于go同步原语Phaser Barrier的资料请关注脚本之家其它相关文章！

最新评论