Golang 并发模式

Go 为并发而生。在使用 Go 编写并发程序时，我们应该熟悉常见的并发模式。虽然业务开发中常用的可能只有那么一两种，但还是有必要了解一下，因为面试可能会被问到。

Go 并发模式指的是对并发协程的管理方式，根据不同的业务场景要求，大概可分为如下几种。

• 全部返回
• 出错及时返回
• 最早成功返回

1.全部返回

全部返回指的是调用下游接口不管失败还是成功，需要等待所有的接口执行完毕。

这种应该是最常见的并发模式，一般使用 Go 官方提供的包 errgroup 便可轻松完成。

假设有三个下游接口需要被调用，这里用三个函数来模拟，并给出不同的耗时。

func api1() (int, error) {
	time.Sleep(time.Second)
	return 1, nil
}

func api2() (int, error) {
	time.Sleep(2*time.Second)
	return 2, nil
}

func api3() (int, error) {
	time.Sleep(3*time.Second)
	return 3, nil
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

使用 errgroup 完成并发调用，并等待所有接口返回。

package main

import (
	"fmt"
	"time"

	"golang.org/x/sync/errgroup"
)

func main() {
	var eg errgroup.Group
	var ret1, ret2, ret3 int
	now := time.Now()
	eg.Go(func() error {
		var err error
		ret1, err = api1()
		return err
	})
	eg.Go(func() error {
		var err error
		ret2, err = api2()
		return err
	})
	eg.Go(func() error {
		var err error
		ret3, err = api3()
		return err
	})
	err := eg.Wait()
	cost := time.Since(now)
	fmt.Printf("err:%v cost:%v ret1:%v ret2:%v ret3:%v\n", err, cost, ret1, ret2, ret3)
}
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

运行输出：

err: cost:3.0012229ss ret1:1 ret2:2 ret3:3
1

通过耗时 cost 为 3s 可见，并发调用下游接口的耗时大约等于其中耗时最久的接口 api3 的耗时。

2.出错及时返回

如果所有的接口都需要成功，业务逻辑上才算成功。那么，当有一个接口返回失败时，其他接口无需再继续等待，即出现错误需及时返回。

还是以三个函数模拟下游被调的接口，假设其中接口 api1 调用发生了失败。

func api1() (int, error) {
	time.Sleep(time.Second)
	return 0, errors.New("api1 failed")
}
1
2
3
4

我们可以借助 select 与 channel 完成对一组协程的并发控制。

package main

import (
	"errors"
	"fmt"
	"time"
)

func main() {
	errchan := make(chan error, 3)
	retchan := make(chan struct{}, 3)
	var ret1, ret2, ret3 int
	now := time.Now()
	go func() {
		var err error
		ret1, err = api1()
		if err != nil {
			errchan <- err
			return
		}
		retchan <- struct{}{}
	}()
	go func() {
		var err error
		ret2, err = api2()
		if err != nil {
			errchan <- err
			return
		}
		retchan <- struct{}{}
	}()
	go func() {
		var err error
		ret3, err = api3()
		if err != nil {
			errchan <- err
			return
		}
		retchan <- struct{}{}
	}()

	// 阻塞等待被调接口出错或全部成功。
	var err error
LOOP:
	for {
		select {
		case err = <-errchan:
			break LOOP
		default:
			if len(retchan) == 3 {
				break
			}
		}
	}
	cost := time.Since(now)
	fmt.Printf("err:%v cost:%vs ret1:%v ret2:%v ret3:%v\n", err, cost, ret1, ret2, ret3)
}
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57

运行输出：

err:api1 failed cost:1.0055006ss ret1:0 ret2:0 ret3:0
1

通过耗时 cost 为 1s 可见，并发调用下游接口，当接口 api1 失败时，不再继续等待其他接口的返回。

3.最早成功返回

如果并发调用多个接口时，只要有一个接口成功返回，其他接口无需再继续等待。即以最早成功返回的那个接口的结果为准，不再关心其他接口的返回。

我们还是假设接口 api1 的耗时最短，但是发生了失败。

func api1() (int, error) {
	time.Sleep(time.Second)
	return 0, errors.New("api1 failed")
}
1
2
3
4

接着我们要继续等待另外两个接口。因为接口 api2 比 api3 耗时短，且成功返回了，所以我们以 api2 返回的结果为准。

func api2() (int, error) {
	time.Sleep(2*time.Second)
	return 2, nil
}
1
2
3
4

我们使用 channel 接收被调接口的结束信号。

package main

import (
	"fmt"
	"time"
)

func main() {
	errchan := make(chan error, 3)
	var ret1, ret2, ret3 int
	now := time.Now()
	go func() {
		var err error
		ret1, err = api1()
		errchan <- err
	}()
	go func() {
		var err error
		ret2, err = api2()
		errchan <- err
	}()
	go func() {
		var err error
		ret3, err = api3()
		errchan <- err
	}()

	var retnum int
	var err error

	// 阻塞等待直至有接口成功返回或全部结束。
LOOP:
	for {
		select {
		case e := <-errchan:
			retnum++
			if e != nil {
				err = e
			}
			if e == nil || retnum == 3 {
				break LOOP
			}
		}
	}
	cost := time.Since(now)
	fmt.Printf("err:%v cost:%vs ret1:%v ret2:%v ret3:%v\n", err, cost, ret1, ret2, ret3)
}
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
44
45
46
47

运行输出：

err:api1 failed cost:2.0001894ss ret1:0 ret2:2 ret3:0
1

通过耗时 cost 为 2s 可见，并发调用下游接口，当接口 api1 失败时，继续等待其他接口。当 api2 成功返回后，则直接结束主协程的阻塞。

4.小结

本文列举了不同业务场景下常见的并发协程管理方式：

• 全部返回
• 出错及时返回
• 最早成功返回

当然还有其他的并发模式，比如生产者消费者模型、发布订阅模型和控制并发数等，具体场景，具体分析，本文不再赘述。

我们可以借助 Go 为我们提供的一系列的同步原语完成对一组协程的控制。比如 sync 包下的 Mutex、RWMutex、WaitGroup、Once、Cond、Pool、Map，以及抽象层级更高的 channel、select、Context 等。除了标准库中提供的同步原语之外，Go 语言还在子仓库 sync 中提供了三种扩展原语 errgroup、semaphore 与 singleflight。

---

参考文献

1.6 常见的并发模式 - Go语言高级编程
Go 语言并发编程、同步原语与锁