极客兔兔

控制协程(goroutine)的并发数量

源代码/数据集已上传到 Github - high-performance-go

high performance go - concurrent programming

1 并发过高导致程序崩溃

我们首先看一个非常简单的例子：

1
2
3
4
5
6
7
8
9
10
11
12
func main() {
	var wg sync.WaitGroup
	for i := 0; i < math.MaxInt32; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			fmt.Println(i)
			time.Sleep(time.Second)
		}(i)
	}
	wg.Wait()
}

这个例子实现了 math.MaxInt32 个协程的并发，约 2^31 = 2 亿个，每个协程内部几乎没有做什么事情。正常的情况下呢，这个程序会乱序输出 1 -> 2^31 个数字。

那实际运行的结果是怎么样的呢？

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
$ go run main.go
...
150577
150578
panic: too many concurrent operations on a single file or socket (max 1048575)

goroutine 1199236 [running]:
internal/poll.(*fdMutex).rwlock(0xc0000620c0, 0x0, 0xc0000781b0)
        /usr/local/go/src/internal/poll/fd_mutex.go:147 +0x13f
internal/poll.(*FD).writeLock(...)
        /usr/local/go/src/internal/poll/fd_mutex.go:239
internal/poll.(*FD).Write(0xc0000620c0, 0xc125ccd6e0, 0x11, 0x20, 0x0, 0x0, 0x0)
        /usr/local/go/src/internal/poll/fd_unix.go:255 +0x5e
fmt.Fprintf(0x10ed3e0, 0xc00000e018, 0x10d3024, 0xc, 0xc0e69b87b0, 0x1, 0x1, 0x11, 0x0, 0x0)
        /usr/local/go/src/fmt/print.go:205 +0xa5
fmt.Printf(...)
        /usr/local/go/src/fmt/print.go:213
main.main.func1(0xc0000180b0, 0x124c31)
...

运行的结果是程序直接崩溃了，关键的报错信息是：

1
panic: too many concurrent operations on a single file or socket (max 1048575)

对单个 file/socket 的并发操作个数超过了系统上限，这个报错是 fmt.Printf 函数引起的，fmt.Printf 将格式化后的字符串打印到屏幕，即标准输出。在 linux 系统中，标准输出也可以视为文件，内核(kernel)利用文件描述符(file descriptor)来访问文件，标准输出的文件描述符为 1，错误输出文件描述符为 2，标准输入的文件描述符为 0。

简而言之，系统的资源被耗尽了。

那如果我们将 fmt.Printf 这行代码去掉呢？那程序很可能会因为内存不足而崩溃。这一点更好理解，每个协程至少需要消耗 2KB 的空间，那么假设计算机的内存是 2GB，那么至多允许 2GB/2KB = 1M 个协程同时存在。那如果协程中还存在着其他需要分配内存的操作，那么允许并发执行的协程将会数量级地减少。

2 如何解决

不同的应用程序，消耗的资源是不一样的。比较推荐的方式的是：应用程序来主动限制并发的协程数量。

2.1 利用 channel 的缓存区

可以利用信道 channel 的缓冲区大小来实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// main_chan.go
func main() {
	var wg sync.WaitGroup
	ch := make(chan struct{}, 3)
	for i := 0; i < 10; i++ {
		ch <- struct{}{}
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			log.Println(i)
			time.Sleep(time.Second)
			<-ch
		}(i)
	}
	wg.Wait()
}
  • make(chan struct{}, 3) 创建缓冲区大小为 3 的 channel，在没有被接收的情况下，至多发送 3 个消息则被阻塞。
  • 开启协程前，调用 ch <- struct{}{}，若缓存区满，则阻塞。
  • 协程任务结束，调用 <-ch 释放缓冲区。
  • sync.WaitGroup 并不是必须的，例如 http 服务，每个请求天然是并发的，此时使用 channel 控制并发处理的任务数量，就不需要 sync.WaitGroup。

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
$ go run main_chan.go
2020/12/21 00:48:28 2
2020/12/21 00:48:28 0
2020/12/21 00:48:28 1
2020/12/21 00:48:29 3
2020/12/21 00:48:29 4
2020/12/21 00:48:29 5
2020/12/21 00:48:30 6
2020/12/21 00:48:30 7
2020/12/21 00:48:30 8
2020/12/21 00:48:31 9

从日志中可以很容易看到，每秒钟只并发执行了 3 个任务，达到了协程并发控制的目的。

2.2 利用第三方库

目前有很多第三方库实现了协程池，可以很方便地用来控制协程的并发数量，比较受欢迎的有：

以 tunny 举例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
package main

import (
	"log"
	"time"

	"github.com/Jeffail/tunny"
)

func main() {
	pool := tunny.NewFunc(3, func(i interface{}) interface{} {
		log.Println(i)
		time.Sleep(time.Second)
		return nil
	})
	defer pool.Close()

	for i := 0; i < 10; i++ {
		go pool.Process(i)
	}
	time.Sleep(time.Second * 4)
}
  • tunny.NewFunc(3, f) 第一个参数是协程池的大小(poolSize)，第二个参数是协程运行的函数(worker)。
  • pool.Process(i) 将参数 i 传递给协程池定义好的 worker 处理。
  • pool.Close() 关闭协程池。

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
$ go run main_tunny.go
2020/12/21 01:00:21 6
2020/12/21 01:00:21 1
2020/12/21 01:00:21 3
2020/12/21 01:00:22 8
2020/12/21 01:00:22 4
2020/12/21 01:00:22 7
2020/12/21 01:00:23 5
2020/12/21 01:00:23 2
2020/12/21 01:00:23 0
2020/12/21 01:00:24 9

3 调整系统资源的上限

3.1 ulimit

有些场景下，即使我们有效地限制了协程的并发数量，但是仍旧出现了某一类资源不足的问题，例如：

  • too many open files
  • out of memory
  • …

例如分布式编译加速工具，需要解析 gcc 命令以及依赖的源文件和头文件，有些编译命令依赖的头文件可能有上百个，那这个时候即使我们将协程的并发数限制到 1000，也可能会超过进程运行时并发打开的文件句柄数量，但是分布式编译工具，仅将依赖的源文件和头文件分发到远端机器执行，并不会消耗本机的内存和 CPU 资源，因此 1000 个并发并不高，这种情况下，降低并发数会影响编译加速的效率，那能不能增加进程能同时打开的文件句柄数量呢？

操作系统通常会限制同时打开文件数量、栈空间大小等，ulimit -a 可以看到系统当前的设置：

1
2
3
4
5
6
7
8
9
10
$ ulimit -a
-t: cpu time (seconds)              unlimited
-f: file size (blocks)              unlimited
-d: data seg size (kbytes)          unlimited
-s: stack size (kbytes)             8192
-c: core file size (blocks)         0
-v: address space (kbytes)          unlimited
-l: locked-in-memory size (kbytes)  unlimited
-u: processes                       1418
-n: file descriptors                12800

我们可以使用 ulimit -n 999999，将同时打开的文件句柄数量调整为 999999 来解决这个问题，其他的参数也可以按需调整。

3.2 虚拟内存(virtual memory)

虚拟内存是一项非常常见的技术了，即在内存不足时，将磁盘映射为内存使用，比如 linux 下的交换分区(swap space)。

在 linux 上创建并使用交换分区是一件非常简单的事情：

1
2
3
4
5
sudo fallocate -l 20G /mnt/.swapfile # 创建 20G 空文件
sudo mkswap /mnt/.swapfile    # 转换为交换分区文件
sudo chmod 600 /mnt/.swapfile # 修改权限为 600
sudo swapon /mnt/.swapfile    # 激活交换分区
free -m # 查看当前内存使用情况(包括交换分区)

关闭交换分区也非常简单：

1
2
sudo swapoff /mnt/.swapfile
rm -rf /mnt/.swapfile

磁盘的 I/O 读写性能和内存条相差是非常大的，例如 DDR3 的内存条读写速率很容易达到 20GB/s，但是 SSD 固态硬盘的读写性能通常只能达到 0.5GB/s，相差 40倍之多。因此，使用虚拟内存技术将硬盘映射为内存使用，显然会对性能产生一定的影响。如果应用程序只是在较短的时间内需要较大的内存，那么虚拟内存能够有效避免 out of memory 的问题。如果应用程序长期高频度读写大量内存，那么虚拟内存对性能的影响就比较明显了。

附 推荐与参考

edit this page last updated at 2023-11-15

赞赏支持

请我吃胡萝卜 =^_^=

i ali

支付宝

i wechat

微信

Big Image