并發(fā)控制模式：orDone的兩種實(shí)現(xiàn)

文章目錄

• 方式一 遞歸

• 方式二 利用反射

• 性能差異

orDone 是一種并發(fā)控制模式，旨在多任務(wù)場(chǎng)景下實(shí)現(xiàn)，有一個(gè)任務(wù)成功返回即立即結(jié)束等待。

今天我們來看下兩種不同的實(shí)現(xiàn)方式：

方式一 遞歸

利用二分法遞歸， 將所有待監(jiān)聽信號(hào)的chan都select起來，

當(dāng)有第一個(gè)chan返回時(shí)，close orDone 來通知讀取方已有第一個(gè)任務(wù)返回

代碼如下比較直觀：

// 傳入多個(gè)并發(fā)chan，返回是否結(jié)束的 orDone chan
func Or(channels ...<-chan interface{}) <-chan interface{} {
 // 只有零個(gè)或者1個(gè)chan
 switch len(channels) {
 case 0:
        // 返回nil， 讓讀取阻塞等待
  return nil
 case 1:
  return channels[0]
 }

 orDone := make(chan interface{})
 go func() {
        // 返回時(shí)利用close做結(jié)束信號(hào)的廣播
  defer close(orDone)

        // 利用select監(jiān)聽第一個(gè)chan的返回
  switch len(channels) {
  case 2: // 直接select
   select {
   case <-channels[0]:
   case <-channels[1]:
   }
  default: // 二分法遞歸處理
   m := len(channels) / 2
   select {
   case <-Or(channels[:m]...):
   case <-Or(channels[m:]...):
   }
  }
 }()

 return orDone
}

方式二 利用反射

這里要用到reflect.SelectCase, 他可以描述一種select的case, 來指明其接受的是chan的讀取或發(fā)送

type SelectCase struct {
 Dir  SelectDir // direction of case
 Chan Value     // channel to use (for send or receive)
 Send Value     // value to send (for send)
}

有了這個(gè)，就可以之間遍歷，不用遞歸來實(shí)現(xiàn)有限的select case構(gòu)造

最后用reflect.Select(cases)監(jiān)聽信號(hào)就可以了，代碼如下：

func OrInReflect(channels ...<-chan interface{}) <-chan interface{} {
 // 只有0個(gè)或者1個(gè)
 switch len(channels) {
 case 0:
  return nil
 case 1:
  return channels[0]
 }

 orDone := make(chan interface{})
 go func() {
  defer close(orDone)
  // 利用反射構(gòu)建SelectCase，這里是讀取
  var cases []reflect.SelectCase
  for _, c := range channels {
   cases = append(cases, reflect.SelectCase{
    Dir:  reflect.SelectRecv,
    Chan: reflect.ValueOf(c),
   })
  }

  // 隨機(jī)選擇一個(gè)可用的case
  reflect.Select(cases)
 }()

 return orDone
}

性能差異

這兩種都可以支持大量chan的信號(hào)監(jiān)聽，那性能差異大么

雖說遞歸開銷肯定不小，反射也不一定效率高，拿個(gè)壓測(cè)來試試吧

先構(gòu)造一下大量并發(fā)chan

func repeat(
 done <-chan interface{},
    // 外部傳入done控制是否結(jié)束
 values ...interface{},
) <-chan interface{} {
 valueStream := make(chan interface{})
 go func() {
        // 返回時(shí)釋放
  defer close(valueStream)
  for {
   for _, v := range values {
    select {
    case <-done:
     return
    case valueStream <- v:
    }
   }
  }
 }()
 return valueStream
}

然后壓測(cè)

func BenchmarkOr(b *testing.B) {
 done := make(chan interface{})
 defer close(done)
 num := 100
 streams := make([]<-chan interface{}, num)
 for i := range streams {
  streams[i] = repeat(done, []int{1, 2, 3})
 }
 b.ResetTimer()
 for i := 0; i < b.N; i++ {
  <-Or(streams...)
 }
}

func BenchmarkOrInReflect(b *testing.B) {
 // 代碼類似
}

跑了下結(jié)果如下：

goos: darwin
goarch: amd64
pkg: github.com/NewbMiao/Dig101-Go/concurrency/channel/schedule/orDone
BenchmarkOr-12                 31815      38136 ns/op     9551 B/op       99 allocs/op
BenchmarkOrInReflect-12        55797      21755 ns/op    25232 B/op      112 allocs/op

可以看出，大量并發(fā)chan場(chǎng)景下， 反射使用內(nèi)存更多些，但速度更快。