Phaser 同步原語

Java 中的 Phaser 是一個同步原語，它可以讓多個線程在某個時刻同步執(zhí)行。它和 Barrier 有點類似，但是它比 Barrier 更加靈活。

舉一個例子哈，比如足球迷特別喜歡的歐洲冠軍聯(lián)賽，它的賽制就分為多個階段：

歐洲冠軍聯(lián)賽由歐冠資格賽、歐冠附加賽和歐冠正賽三部分組成。 歐冠資格賽，分為預(yù)賽輪（preliminary round）、第一輪資格賽（first qualifying round）、第二輪資格賽（second qualifying round）和第三輪資格賽（third qualifying round）。第三輪資格賽的優(yōu)勝的 10 支球隊進(jìn)入歐冠附加賽，附加賽優(yōu)勝的 6 支球隊（冠軍之路 4 隊、聯(lián)賽之路 2 隊）將和 26 支自動晉級的隊伍一起，參加歐冠小組賽。 歐冠正賽分為小組賽、1/8 決賽、1/4 決賽、半決賽和決賽。

所以歐冠聯(lián)賽分成了多個階段，每一個階段，會有一些球隊參加，等到下一階段，淘汰了一部分球隊，又會有新的球隊加入，每個階段的球隊都會有變化。這種情況非常適合使用 Phaser 來模擬。

Phaser 和 CyclicBarrier對比

Phaser 和 CyclicBarrier的功能非常的相似，都是應(yīng)用于多個參與者多階段處理問題的場景，每個階段都有障礙點，在障礙點需要等待所有的參與者到齊后才能進(jìn)入下一個階段。但是Phaser比CyclicBarrier更加靈活，CyclicBarrier的參與者數(shù)量是固定的，所以初始化CyclicBarrier的時候就需要設(shè)定參與者的數(shù)量，而Phaser的參與者數(shù)量是可以動態(tài)變化的，每個階段完成后參與者可以選擇離開，新的參與者也可以加入進(jìn)來，所以上面歐冠的例子非常使用Phaser來模擬。

一旦一個同步原語的功能不是那么通用，而是面向非常細(xì)分的場景，那么它的使用范圍非常有限，因為大部分場景我們都會使用WaitGroup、channel甚至CyclicBarrier去解決，但是針對參與者需要動態(tài)變化的場景，我們使用Phaser如魚得水，比自己再封裝和實現(xiàn)類似Phaser的功能更方便。正所謂“技多不壓身”，我們多了解一些同步原語，在解決問題的時候就會更加得心應(yīng)手。

Go 標(biāo)準(zhǔn)庫和擴(kuò)展庫中都沒有實現(xiàn)，第三方庫也鮮有實現(xiàn)，但是 Java 中有，我們可以參考 Java 中的實現(xiàn)，自己實現(xiàn)了一個，比如 github.com/smallnest/exp/sync/Phaser[1]，當(dāng)然針對 Java 復(fù)雜的實現(xiàn)做了精簡，不再支持 Phaser 的父子關(guān)系，函數(shù)名也做了簡化，將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，指定參與者的數(shù)量。

  • NewPhaserWithAction：初始化一個 Phaser，指定參與者的數(shù)量，以及每個階段的障礙點到達(dá)后的回調(diào)函數(shù)。

• 動作

  • Arrive：參與者到達(dá)障礙點，但是不等待其他參與者，直接返回當(dāng)前階段。階段的編號從 0 開始，每進(jìn)入一個新的階段，階段編號會自增 1。

  • ArriveAndLeave：參與者到達(dá)障礙點，但是不等待其他參與者，直接返回當(dāng)前階段，并且離開 Phaser。

  • Wait：等待指定的階段，如果指定的階段已經(jīng)完成，直接返回，否則等待指定的階段完成后返回。

  • ArriveAndWait：參與者到達(dá)障礙點，等待其他參與者到達(dá)障礙點，然后返回當(dāng)前階段。

• 加入和離開

  • Join：參與者加入 Phaser，參與者的數(shù)量會自增 1。

  • Leave：參與者離開 Phaser，參與者的數(shù)量會自減 1。

  • BulkJoin：批量加入?yún)⑴c者，參與者的數(shù)量會自增指定的數(shù)量。

• 終止

  • ForceTermination：強(qiáng)制終止 Phaser，所有的參與者都會離開 Phaser。

• 查詢

  • Arrived：返回當(dāng)前階段已經(jīng)到達(dá)障礙點的參與者數(shù)量。

  • Parties：返回當(dāng)前 Phaser 中參與者的數(shù)量。

  • Phase：返回當(dāng)前階段的編號。

  • IsTerminated：返回 Phaser 是否已經(jīng)終止。

解釋代碼邏輯

舉一個例子,我在代碼中加上注釋，來解釋代碼的邏輯：

func TestPhaser_phase(t *testing.T) {
 // 模擬一個有三個階段的場景，其中一個參與者(id=1)在參加完第一階段就離開了
    // 初始化一個Phaser，指定參與者的數(shù)量為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， 第一階段，等待其他參與者到達(dá)
   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， 第二階段，等待其他參與者到達(dá)
   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， 第三階段，等待其他參與者到達(dá)
   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() // 強(qiáng)制終止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區(qū)別的詳細(xì)內(nèi)容，更多關(guān)于go同步原語Phaser Barrier的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論