Go函数并发情况的错误处理

前言

最近遇到了一个很有意思的问题, 感觉值得写一篇博客来记录一下, 也在大家遇到这种问题的时候可以有个参考;
下面这段代码大家都不陌生吧, 一个简单的多go程处理, 大家可以看看有没有什么问题

func handle() {
  var (
    errCh   = make(chan error, 1)
    doneCh  = make(chan struct{})
    records = make([]string, 100)
    gp      = gopool.NewGoPool(10) // go程池, 只允许开10个go程
  )

  for _, v := range records {
    if len(errCh) > 0 {
      break
    }
    gp.Add()
    go func(v string) {
      defer gp.Done()
      // handles
      // check(err)
      errCh <- fmt.Errorf("err")
    }(v)
  }

  go func() {
    gp.Wait()
    doneCh <- struct{}{}
  }()

  select {
  case <-doneCh:
    fmt.Println("done")
  case <-errCh:
    fmt.Println("err")
  }
}

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

问题

其实这段代码里有很大的安全隐患, 列举一下:

1. 如果 for循环不判断 errCh, 在100个循环完之前产生11个err, errCh插入不进去, for循环就会直接死锁, 走不到下面的select;
2. 虽然判断了errCh保证可以终止for循环, 但是其他go程产生错误也会死锁, 并且每次运行这个函数都有可能产生死锁go程, 会产生 Goroutine Leak;
3. 最后select不能完全保证doneCh和errCh 哪个优先监听到;

那么有问题就要有对应的解决思路及方案, 下面给出几个:

方案一 扩充errCh的大小;

点评

最简单粗暴的方案, 但是会造成内存浪费;

实现

   errCh   = make(chan error, len(records))
1

方案二 使用context.WithCancel;

点评

context替换errCh作为监听err, cancel() 可以多次执行, 不会阻塞;
比思路一优雅一些, 但是代码会比较冗余;

实现

func handle() {
  var (
    ctx, cancel = context.WithCancel(context.Background())
    doneCh      = make(chan struct{})
    records     = make([]string, 100)
    gp          = gopool.NewGoPool(10) // go程池, 只允许开10个go程
  )

  go func() {
    for _, v := range records {
      select {
      case <-ctx.Done():
        return
      default:
      }
      gp.Add()
      go func(v string) {
        defer gp.Done()
        // handles
        // check(err)
        cancel()
      }(v)
    }
  }()

  go func() {
    gp.Wait()
    time.Sleep(1 * time.Millisecond) // 优先监听errCh
    select {
    case <-ctx.Done():
      return
    default:
    }
    doneCh <- struct{}{}
  }()

  select {
  case <-doneCh:
    fmt.Println("done")
  case <-ctx.Done():
    fmt.Println("err")
  }
}
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
35
36
37
38
39
40
41
42
43

方案三 使用sync.Once

点评

使用 Once 来限制 插入errCh操作只执行一次;
目前最优雅的思路, 代码改动也最少;

实现

func handle() {
  var (
    errCh   = make(chan error, 1)
    doneCh  = make(chan struct{})
    records = make([]string, 100)
    gp      = gopool.NewGoPool(10) // go程池, 只允许开10个go程
    _once   = new(sync.Once)
  )

  for _, v := range records {
    if len(errCh) > 0 {
      break
    }
    gp.Add()
    go func(v string) {
      defer gp.Done()
      // handles
      // check(err)
      _once.Do(func() {
        errCh <- fmt.Errorf("err")
      })
    }(v)
  }

  go func() {
    gp.Wait()
    time.Sleep(1 * time.Millisecond) // 优先监听errCh
    doneCh <- struct{}{}
  }()

  select {
  case <-doneCh:
    fmt.Println("done")
  case <-errCh:
    fmt.Println("err")
  }
}
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
35
36
37

结束语

大家如果有更好的思路/方案可以在评论区/私信给我, 共同学习共同进步;

我自己写了一个Go开箱即用的开源项目, 里面封装了常用的一些组件, git clone下来就可以直接进行API开发, 有兴趣的可以给个Star, 会一直持续维护;

项目地址 https://github.com/shixiaofeia/fly