Go语言并发如何使用才更加高效

Posted Golang语言社区

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Go语言并发如何使用才更加高效相关的知识,希望对你有一定的参考价值。



This browser does not support music or audio playback. Please play it in WeChat or another browser. Go语言并发如何使用才更加高效 Go语言并发如何使用才更加高效


很多刚开始使用 Go 语言开发的人都很喜欢使用并发特性,而没有考虑并发是否真正能解决他们的问题。了解goroutine的生命期时再创建goroutine在 Go 语言中,开发者习惯将并发内容与 goroutine 一一对应地创建 goroutine。开发者很少会考虑 goroutine 在什么时候能退出和控制 goroutine 生命期,这就会造成 goroutine 失控的情况。下面来看一段代码。

失控的 goroutine:

 1package main
2
3import (
4    "fmt"
5    "runtime"
6)
7
8// 一段耗时的计算函数
9func consumer(ch chan int) {
10
11    // 无限获取数据的循环
12    for {
13
14        // 从通道获取数据
15        data := <-ch
16
17        // 打印数据
18        fmt.Println(data)
19    }
20
21}
22
23func main() {
24
25    // 创建一个传递数据用的通道
26    ch := make(chan int)
27
28    for {
29
30        // 空变量, 什么也不做
31        var dummy string
32
33        // 获取输入, 模拟进程持续运行
34        fmt.Scan(&dummy)
35
36        // 启动并发执行consumer()函数
37        go consumer(ch)
38
39        // 输出现在的goroutine数量
40        fmt.Println("goroutines:", runtime.NumGoroutine())
41    }
42
43}

代码说明如下:

  • 第 9 行,consumer() 函数模拟平时业务中放到 goroutine 中执行的耗时操作。该函数从其他 goroutine 中获取和接收数据或者指令,处理后返回结果。

  • 第 12 行,需要通过无限循环不停地获取数据。

  • 第 15 行,每次从通道中获取数据。

  • 第 18 行,模拟处理完数据后的返回数据。

  • 第 26 行,创建一个整型通道。

  • 第 37 行,启动并发执行 consumer() 函数,并传入 ch 通道。

  • 第 40 行,每启动一个 goroutine,使用 runtime.NumGoroutine 检查进程创建的 goroutine 数量总数。


运行程序,每输入一个字符串+回车,将会创建一个 goroutine,结果如下:

1a
2goroutines: 2
3b
4goroutines: 3
5c
6goroutines: 4

注意,结果中 a、b、c 为通过键盘输入的字符,其他为打印字符。

这个程序实际在模拟一个进程根据需要创建 goroutine 的情况。运行后,问题已经被暴露出来:随着输入的字符串越来越多,goroutine 将会无限制地被创建,但并不会结束。这种情况如果发生在生产环境中,将会造成内存大量分配,最终使进程崩溃。现实的情况也许比这段代码更加隐蔽:也许你设置了一个退出的条件,但是条件永远不会被满足或者触发。

为了避免这种情况,在这个例子中,需要为 consumer() 函数添加合理的退出条件,修改代码后如下:

 1package main
2
3import (
4    "fmt"
5    "runtime"
6)
7
8// 一段耗时的计算函数
9func consumer(ch chan int) {
10
11    // 无限获取数据的循环
12    for {
13
14        // 从通道获取数据
15        data := <-ch
16
17        if data == 0 {
18            break
19        }
20
21        // 打印数据
22        fmt.Println(data)
23    }
24
25    fmt.Println("goroutine exit")
26}
27
28func main() {
29
30    // 传递数据用的通道
31    ch := make(chan int)
32
33    for {
34
35        // 空变量, 什么也不做
36        var dummy string
37
38        // 获取输入, 模拟进程持续运行
39        fmt.Scan(&dummy)
40
41        if dummy == "quit" {
42
43            for i := 0; i < runtime.NumGoroutine()-1; i++ {
44                ch <- 0
45            }
46
47            continue
48        }
49
50        // 启动并发执行consumer()函数
51        go consumer(ch)
52
53        // 输出现在的goroutine数量
54        fmt.Println("goroutines:", runtime.NumGoroutine())
55    }
56}

代码中加粗部分是新添加的代码,具体说明如下:

  • 第 17 行,为无限循环设置退出条件,这里设置 0 为退出。

  • 第 41 行,当命令行输入 quit 时,进入退出处理的流程。

  • 第 43 行,runtime.NumGoroutine 返回一个进程的所有 goroutine 数,main() 的 goroutine 也被算在里面。因此需要扣除 main() 的 goroutine 数。剩下的 goroutine 为实际创建的 goroutine 数,对这些 goroutine 进行遍历。

  • 第 44 行,并发开启的 goroutine 都在竞争获取通道中的数据,因此只要知道有多少个 goroutine 需要退出,就给通道里发多少个 0。


修改程序并运行,结果如下:

1a
2goroutines: 2
3b
4goroutines: 3
5quit
6goroutine exit
7goroutine exit
8c
9goroutines: 2

避免在不必要的地方使用通道通道(channel)和 map、切片一样,也是由 Go 源码编写而成。为了保证两个 goroutine 并发访问的安全性,通道也需要做一些锁操作,因此通道其实并不比锁高效。

下面的例子展示套接字的接收和并发管理。对于 TCP 来说,一般是接收过程创建 goroutine 并发处理。当套接字结束时,就要正常退出这些 goroutine。


下面是对各个部分的详细分析。
1) 套接字接收部分套接字在连接后,就需要不停地接收数据,代码如下:

 1// 套接字接收过程
2func socketRecv(conn net.Conn, exitChan chan string) {
3
4// 创建一个接收的缓冲
5    buff := make([]byte1024)
6
7    // 不停地接收数据
8    for {
9
10        // 从套接字中读取数据
11        _, err := conn.Read(buff)
12
13        // 需要结束接收, 退出循环
14        if err != nil {
15            break
16        }
17
18    }
19
20    // 函数已经结束, 发送通知
21    exitChan <- "recv exit"
22}

代码说明如下:

  • 第 2 行传入的 net.Conn 是套接字的接口,exitChan 为退出发送同步通道。

  • 第 5 行为套接字的接收数据创建一个缓冲。

  • 第 8 行构建一个接收的循环,不停地接收数据。

  • 第 11 行,从套接字中取出数据。这个例子中,不关注具体接收到的数据,只是关注错误,这里将接收到的字节数做匿名处理。

  • 第 14 行,当套接字调用了 Close 方法时,会触发错误,这时需要结束接收循环。

  • 第 21 行,结束函数时,与函数绑定的 goroutine 会同时结束,此时需要通知 main() 的 goroutine。

2) 连接、关闭、同步 goroutine 主流程部分下面代码中尝试使用套接字的 TCP 协议连接一个网址,连接上后,进行数据接收,等待一段时间后主动关闭套接字,等待套接字所在的 goroutine 自然结束,代码如下:

 1func main() {
2
3    // 连接一个地址
4    conn, err := net.Dial("tcp""www.163.com:80")
5
6    // 发生错误时打印错误退出
7    if err != nil {
8        fmt.Println(err)
9        return
10    }
11
12    // 创建退出通道
13    exit := make(chan string)
14
15    // 并发执行套接字接收
16    go socketRecv(conn, exit)
17
18    // 在接收时, 等待1秒
19    time.Sleep(time.Second)
20
21    // 主动关闭套接字
22    conn.Close()
23
24    // 等待goroutine退出完毕
25    fmt.Println(<-exit)
26}

代码说明如下:

  • 第 4 行,使用 net.Dial 发起 TCP 协议的连接,调用函数就会发送阻塞直到连接超时或者连接完成。

  • 第 7 行,如果连接发生错误,将会打印错误并退出。

  • 第 13 行,创建一个通道用于退出信号同步,这个通道会在接收用的 goroutine 中使用。

  • 第 16 行,并发执行接收函数,传入套接字和用于退出通知的通道。

  • 第 19 行,接收需要一个过程,使用 time.Sleep() 等待一段时间。

  • 第 22 行,主动关闭套接字,此时会触发套接字接收错误。

  • 第 25 行,从 exit 通道接收退出数据,也就是等待接收 goroutine 结束。


在这个例子中,goroutine 退出使用通道来通知,这种做法可以解决问题,但是实际上通道中的数据并没有完全使用。3) 优化:使用等待组替代通道简化同步通道的内部实现代码在 Go 语言开发包的 src/runtime/chan.go 中,经过分析后大概了解到通道也是用常见的互斥量等进行同步。因此通道虽然是一个语言级特性,但也不是被神化的特性,通道的运行和使用都要比传统互斥量、等待组(sync.WaitGroup)有一定的消耗。

所以在这个例子中,更建议使用等待组来实现同步,调整后的代码如下:

 1package main
2
3import (
4    "fmt"
5    "net"
6    "sync"
7    "time"
8)
9
10// 套接字接收过程
11func socketRecv(conn net.Conn, wg *sync.WaitGroup) {
12
13    // 创建一个接收的缓冲
14    buff := make([]byte1024)
15
16    // 不停地接收数据
17    for {
18
19        // 从套接字中读取数据
20        _, err := conn.Read(buff)
21
22        // 需要结束接收, 退出循环
23        if err != nil {
24            break
25        }
26
27    }
28
29    // 函数已经结束, 发送通知
30    wg.Done()
31}
32
33func main() {
34
35    // 连接一个地址
36    conn, err := net.Dial("tcp""www.163.com:80")
37
38    // 发生错误时打印错误退出
39    if err != nil {
40        fmt.Println(err)
41        return
42    }
43
44    // 退出通道
45    var wg sync.WaitGroup
46
47    // 添加一个任务
48    wg.Add(1)
49
50    // 并发执行接收套接字
51    go socketRecv(conn, &wg)
52
53    // 在接收时, 等待1秒
54    time.Sleep(time.Second)
55
56    // 主动关闭套接字
57    conn.Close()
58
59    // 等待goroutine退出完毕
60    wg.Wait()
61    fmt.Println("recv done")
62}


调整后的代码说明如下:

  • 第 45 行,声明退出同步用的等待组。

  • 第 48 行,为等待组的计数器加 1,表示需要完成一个任务。

  • 第 51 行,将等待组的指针传入接收函数。

  • 第 60 行,等待等待组的完成,完成后打印提示。

  • 第 30 行,接收完成后,使用 wg.Done() 方法将等待组计数器减一。

以上是关于Go语言并发如何使用才更加高效的主要内容,如果未能解决你的问题,请参考以下文章

使用Go语言实现高效的并发编程

使用Go语言实现高效的并发编程

go语言简介

[GO专栏-2]Go语言的设计哲学

15. Go 语言“避坑”与技巧

谈谈 Golang 并发编程