go►dig101

or-channel模式的两种实现

文章目录

  1. 1. 方式一 递归
  2. 2. 方式二 利用反射
  3. 3. 性能差异

or channel模式是一种并发控制模式，旨在多任务场景下实现，有一个任务成功返回即立即结束等待。

今天我们来看下两种不同的实现方式：

方式一 递归

利用二分法递， 将所有待监听信号的chan都select起来，

当有第一个chan返回时，close orDone 来通知读取方已有第一个任务返回

代码如下比较直观：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
// 传入多个并发chan，返回是否结束的 orDone chan
func Or(channels ...<-chan interface{}) <-chan interface{} {
 // 只有零个或者1个chan
 switch len(channels) {
 case 0:
        // 返回nil， 让读取阻塞等待
  return nil
 case 1:
  return channels[0]
 }

 orDone := make(chan interface{})
 go func() {
        // 返回时利用close做结束信号的广播
  defer close(orDone)

        // 利用select监听第一个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的读取或发送

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

有了这个，就可以之间遍历，不用递归来实现有限的select case构造

最后用reflect.Select(cases)监听信号就可以了，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
func OrInReflect(channels ...<-chan interface{}) <-chan interface{} {
 // 只有0个或者1个
 switch len(channels) {
 case 0:
  return nil
 case 1:
  return channels[0]
 }

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

  // 随机选择一个可用的case
  reflect.Select(cases)
 }()

 return orDone
}

性能差异

这两种都可以支持大量chan的信号监听，那性能差异大么

虽说递归开销肯定不小，反射也不一定效率高，拿个压测来试试吧

先构造一下chan

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func AsStream(done <-chan struct{}, values ...interface{}) <-chan interface{} {
 s := make(chan interface{})
 go func() {
  // 退出时关闭chan             
  defer close(s)
  for _, v := range values {
   select {
   case <-done:
    return
   case s <- v:
   }
  }
 }()
 return s
}

然后压测

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
var funcs = []struct {
 name string
 f    func(...<-chan interface{}) <-chan interface{}
}{
 {"reflection", OrInReflect},
 {"recursion", OrRecur},
}
func BenchmarkOr(b *testing.B) {
 for _, f := range funcs {
  for n := 8; n <= 1024; n *= 2 {
   b.Run(fmt.Sprintf("%s/%d", f.name, n), func(b *testing.B) {
    for i := 0; i < b.N; i++ {
     b.StopTimer()
     done := make(chan struct{})
     defer close(done)
     streams := make([]<-chan interface{}, n)
     for i := range streams {
      streams[i] = AsStream(done, []interface{}{1})
     }
     b.StartTimer()
     <-f.f(streams...)
    }
   })
  }
 }
}

跑了下结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
goos: darwin
goarch: amd64
pkg: github.com/NewbMiao/Dig101-Go/concurrency/channel/schedule/or
cpu: Intel(R) Core(TM) i7-8750H CPU @ 2.20GHz
BenchmarkOr/reflection/8-12                    60044             22924 ns/op            2839 B/op         28 allocs/op
BenchmarkOr/reflection/16-12                   82941             27764 ns/op            4684 B/op         41 allocs/op
BenchmarkOr/reflection/32-12                   56890             40326 ns/op            8397 B/op         66 allocs/op
BenchmarkOr/reflection/64-12                   33350             90301 ns/op           16485 B/op        116 allocs/op
BenchmarkOr/reflection/128-12                  17476            109545 ns/op           34300 B/op        230 allocs/op
BenchmarkOr/reflection/256-12                   8155            257080 ns/op           68398 B/op        443 allocs/op
BenchmarkOr/reflection/512-12                   4018            429550 ns/op          134260 B/op        842 allocs/op
BenchmarkOr/reflection/1024-12                  2131            890946 ns/op          266877 B/op       1648 allocs/op
BenchmarkOr/recursion/8-12                    186949              6770 ns/op            1190 B/op         12 allocs/op
BenchmarkOr/recursion/16-12                   127618             10651 ns/op            2048 B/op         21 allocs/op
BenchmarkOr/recursion/32-12                    83200             24578 ns/op            3405 B/op         35 allocs/op
BenchmarkOr/recursion/64-12                    69890             33589 ns/op            5162 B/op         53 allocs/op
BenchmarkOr/recursion/128-12                   32719             58391 ns/op            8301 B/op         86 allocs/op
BenchmarkOr/recursion/256-12                   10000            162016 ns/op           13487 B/op        140 allocs/op
BenchmarkOr/recursion/512-12                    7225            283879 ns/op           24199 B/op        252 allocs/op
BenchmarkOr/recursion/1024-12                   3112            645105 ns/op           50744 B/op        528 allocs/op

压测结果如图所示
orBenchmark

可以看出，大量并发chan场景下， 递归效率更好一些。

如有疑问，请文末留言交流或邮件：newbvirgil@gmail.com 本文链接 : https://newbmiao.github.io/2021/08/19/2-way-of-or-done-pattern.html
分享到 次阅读   |  最近更新 ：