Go缓冲通道和工作池

什么是缓冲通道？

我们在上一篇教程中讨论的所有通道基本上都是无缓冲的。正如我们在通道教程中详细讨论的那样，向无缓冲通道的发送和接收是阻塞的。

可以创建带有缓冲区的通道。仅当缓冲区已满时，才会阻止发送到缓冲通道。类似地，只有当缓冲区为空时，来自缓冲通道的接收才会被阻止。

make可以通过将附加容量参数传递给指定缓冲区大小的函数来创建缓冲通道。

ch := make(chan type, capacity)  
1

对于具有缓冲区的通道，上述语法中的容量应大于 0。无缓冲通道的容量默认为 0，因此我们在上一教程中创建通道时省略了容量参数。

让我们编写一些代码并创建一个缓冲通道。

例子

package main

import (  
    "fmt"
)


func main() {  
    ch := make(chan string, 2)
    ch <- "naveen"
    ch <- "paul"
    fmt.Println(<- ch)
    fmt.Println(<- ch)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

Run program in playground

在上面的程序中，第 9 行。 我们创建一个容量为2的缓冲通道。由于通道的容量为2，因此可以向通道写入2个字符串而不会被阻塞。我们在第10和11行向通道写入 2 个字符串。并且通道不阻塞。我们读取 2 个字符串。该程序打印，

naveen  
paul  
1
2

另一个例子

让我们再看一个缓冲通道的示例，其中通道的值被写入并发 Goroutine 中并从主 Goroutine 读取。

这个例子将帮助我们更好地理解何时写入缓冲通道块。

package main

import (  
    "fmt"
    "time"
)

func write(ch chan int) {  
    for i := 0; i < 5; i++ {
        ch <- i
        fmt.Println("successfully wrote", i, "to ch")
    }
    close(ch)
}
func main() {  
    ch := make(chan int, 2)
    go write(ch)
    time.Sleep(2 * time.Second)
    for v := range ch {
        fmt.Println("read value", v,"from ch")
        time.Sleep(2 * time.Second)

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

Run program in playground

在上面的程序中，第 16 行创建了一个ch容量缓冲通道。第 17行main中调用Goroutine write并传递ch。然后主 Goroutine 休眠 2 秒。在此期间，Goroutine write正在并发运行。Goroutinewrite有一个for循环，将 0 到 4 之间的数字写入ch通道。该缓冲通道的容量为2，因此写入Goroutine将能够立即将值0h和1写入通道ch，然后它会阻塞，直到从通道读取至少一个值ch。所以这个程序会立即打印以下两行。

successfully wrote 0 to ch  
successfully wrote 1 to ch  
1
2

打印以上两行后， Goroutinewrite中对通道ch的写入将被阻塞，直到有人从ch通道中读取。由于主 Goroutine 在开始从通道读取之前会休眠 2 秒，因此程序在接下来的 2 秒内不会打印任何内容。Goroutinemain在 2 秒后唤醒，并开始ch使用for range第 19行中的循环从通道读取数据。打印读取到的值，然后再次休眠2秒，如此循环，直到ch关闭。所以程序会在 2 秒后打印以下几行：

read value 0 from ch  
successfully wrote 2 to ch  
1
2

这将持续下去，直到所有值都写入通道并通道关闭。最终的输出是，

successfully wrote 0 to ch  
successfully wrote 1 to ch  
read value 0 from ch  
successfully wrote 2 to ch  
read value 1 from ch  
successfully wrote 3 to ch  
read value 2 from ch  
successfully wrote 4 to ch  
read value 3 from ch  
read value 4 from ch  
1
2
3
4
5
6
7
8
9
10

死锁

package main

import (  
    "fmt"
)

func main() {  
    ch := make(chan string, 2)
    ch <- "naveen"
    ch <- "paul"
    ch <- "steve"
    fmt.Println(<-ch)
    fmt.Println(<-ch)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

Run program in playground

在上面的程序中，我们将 3 个字符串写入容量为 2 的缓冲通道。当控制到达第 11 行的第三个写入时，由于通道超出其容量，写入被阻止。现在，必须从通道读取数据才能继续写入，但在这种情况下，没有并发例程从该通道读取数据。因此，将会出现死锁，并且程序将在运行时出现紧急情况，并显示以下消息：

fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan send]:  
main.main()  
    /tmp/sandbox091448810/prog.go:11 +0x8d
1
2
3
4
5

关闭缓冲通道

我们已经在上一篇教程中讨论了关闭通道。除了我们在上一教程中学到的内容之外，关闭缓冲通道时还需要考虑一个微妙之处。

可以从已经关闭的缓冲通道读取数据。通道将返回已写入通道的数据，一旦读取完所有数据，将返回通道的零值。

让我们编写一个程序来理解这一点。

package main

import (  
    "fmt"
)

func main() {  
    ch := make(chan int, 5)
    ch <- 5
    ch <- 6
    close(ch)
    n, open := <-ch 
    fmt.Printf("Received: %d, open: %t\n", n, open)
    n, open = <-ch 
    fmt.Printf("Received: %d, open: %t\n", n, open)
    n, open = <-ch 
    fmt.Printf("Received: %d, open: %t\n", n, open)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

Run program in playground

在上面的程序中，我们在第 8 行创建了一个容量为5缓冲通道。 然后我们将5和6写入通道。通道在第11 行之后关闭。即使通道关闭，我们也可以读取已经写入通道的值。这是在第 12和14 行中完成的。我们现在已经完成从通道中读取5和6，并且没有更多数据需要读取。当在第16行再次读取通道时。其实n则是为int的0值，open 则为false

该程序将打印

Received: 5, open: true  
Received: 6, open: true  
Received: 0, open: false  
1
2
3

也可以使用 for range 循环编写相同的程序。

package main

import (  
    "fmt"
)

func main() {  
    ch := make(chan int, 5)
    ch <- 5
    ch <- 6
    close(ch)
    for n := range ch {
        fmt.Println("Received:", n)
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

Run program in playground

该程序将打印，

Received: 5  
Received: 6  
1
2

长度与容量

缓冲通道的容量是通道可以保存的值的数量。这是我们在使用该函数创建缓冲通道时指定的值。

缓冲通道的长度是当前在其中排队的元素的数量。

一个程序会让事情变得清晰😀

package main

import (  
    "fmt"
)

func main() {  
    ch := make(chan string, 3)
    ch <- "naveen"
    ch <- "paul"
    fmt.Println("capacity is", cap(ch))
    fmt.Println("length is", len(ch))
    fmt.Println("read value", <-ch)
    fmt.Println("new length is", len(ch))
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

Run program in playground

在上面的程序中，创建的通道容量为3，即可以容纳 3 个字符串。然后我们将 2 个字符串写入。现在通道中有 2 个字符串在其中排队，因此其长度为2。在我们从通道读取一个字符串。现在通道中只有一个字符串在排队，因此其长度变为1。该程序将打印，

capacity is 3  
length is 2  
read value naveen  
new length is 1
1
2
3
4

WaitGroup

本教程的下一部分是关于工作池的。要了解Worker Pool，我们首先需要了解WaitGroup的实现中会用到它。

WaitGroup 用于等待 Goroutine 集合完成执行。控制被阻塞，直到所有 Goroutines 执行完毕。

假设我们有 3 个并发执行的 Goroutine，这些 Goroutine 是从Goroutine main中衍生出来的。Goroutinemain需要等待其他 3 个 Goroutine 完成才能终止。这可以使用 WaitGroup 来完成。

让我们停止理论并立即编写一些代码😀

package main

import (  
    "fmt"
    "sync"
    "time"
)

func process(i int, wg *sync.WaitGroup) {  
    fmt.Println("started Goroutine ", i)
    time.Sleep(2 * time.Second)
    fmt.Printf("Goroutine %d ended\n", i)
    wg.Done()
}

func main() {  
    no := 3
    var wg sync.WaitGroup
    for i := 0; i < no; i++ {
        wg.Add(1)
        go process(i, &wg)
    }
    wg.Wait()
    fmt.Println("All go routines finished executing")
}
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

Run program in playground

WaitGroup是一个结构类型，我们在第 18 行创建一个WaitGroup类型的零值变量wg。WaitGroup工作方式是使用计数器。当我们调用Add并传递给它时int，WaitGroup的计数器会根据传递给的值递增。递减计数器的方法是调用Done()。该Wait()方法会阻塞Goroutine调用它的地方，直到计数器变为零。

在上面的程序中，我们调用wg.Add(1) 在for循环内迭代 3 次。所以计数器现在变成 3。for循环还生成 3 个Goroutinesprocess，然后wg.Wait()在第 23行调用。 使mainGoroutine 等待，直到计数器变为零。

wg传递第 21 行的指针很重要。.如果不传递指针，那么每个Goroutine都会有自己的副本，WaitGroup在执行完成时main不会收到通知。

该程序输出。

started Goroutine  2  
started Goroutine  0  
started Goroutine  1  
Goroutine 0 ended  
Goroutine 2 ended  
Goroutine 1 ended  
All go routines finished executing  
1
2
3
4
5
6
7

您的输出可能与我的不同，因为 Goroutines 的执行顺序可能不同

工作池

缓冲通道的重要用途之一是工作池的实现。

一般来说，工作池是等待分配任务的线程的集合。一旦他们完成分配的任务，他们就会再次为下一个任务做好准备。

我们将使用缓冲通道实现一个工作池。

我们的工作池将执行查找输入数字的数字之和的任务。例如，如果传递 234，则输出将为 9 (2 + 3 + 4)。工作池的输入将是伪随机整数列表。

以下是我们工作池的核心功能

• 创建一个 Goroutines 池，监听输入缓冲通道，等待分配作业
• 将作业添加到输入缓冲通道
• 作业完成后将结果写入输出缓冲通道
• 从输出缓冲通道读取并打印结果

我们将一步一步地编写这个程序，以使其更容易理解。

第一步是创建代表作业和结果的结构。

type Job struct {  
    id       int
    randomno int
}
type Result struct {  
    job         Job
    sumofdigits int
}
1
2
3
4
5
6
7
8

每个Job结构体都有 id和 randomno，必须计算它们各个数字的总和。

该Result结构有一个job字段，sumofdigits该字段是其在字段中保存结果（各个数字的总和）的作业 。

下一步是创建用于接收作业和写入输出的缓冲通道。

var jobs = make(chan Job, 10)  
var results = make(chan Result, 10)  
1
2

Worker Goroutine 监听jobs缓冲通道上的新任务。任务完成后，结果将写入results缓冲通道。

下面的digits函数实际工作是查找整数的各个数字的总和并返回它。我们将给这个函数添加 2 秒的睡眠，只是为了模拟这个函数需要一些时间来计算结果。

func digits(number int) int {  
    sum := 0
    no := number
    for no != 0 {
        digit := no % 10
        sum += digit
        no /= 10
    }
    time.Sleep(2 * time.Second)
    return sum
}
1
2
3
4
5
6
7
8
9
10
11

接下来，我们将编写一个创建 Worker Goroutine 的函数。

func worker(wg *sync.WaitGroup) {  
    for job := range jobs {
        output := Result{job, digits(job.randomno)}
        results <- output
    }
    wg.Done()
}
1
2
3
4
5
6
7

上面的函数创建了一个工作线程，它从通道读取数据，循环jobs通道然后将结果写入缓冲通道。该worker函数采用 WaitGroup作为参数，当所有操作完成后，它将调用wg.Done()方法。

该createWorkerPool函数将创建一个工作池。

func createWorkerPool(noOfWorkers int) {  
    var wg sync.WaitGroup
    for i := 0; i < noOfWorkers; i++ {
        wg.Add(1)
        go worker(&wg)
    }
    wg.Wait()
    close(results)
}
1
2
3
4
5
6
7
8
9

上面的函数将要创建的工作人员数量作为参数。它在创建之前调用 WaitGroup 计数器。然后，它通过将 WaitGroup 的指针传递wg给该worker函数来创建工作 Goroutine。创建所需的 Worker Goroutine 后，它会通过调用wg.Wait() 来等待所有 Goroutine 完成执行。当所有 Goroutine 完成执行后，它会关闭results通道，因为所有 Goroutines 都已完成执行，并且没有其他人会进一步写入通道results。

现在我们已经准备好了工作池，让我们继续编写将作业分配给工作人员的函数。

func allocate(noOfJobs int) {  
    for i := 0; i < noOfJobs; i++ {
        randomno := rand.Intn(999)
        job := Job{i, randomno}
        jobs <- job
    }
    close(jobs)
}
1
2
3
4
5
6
7
8

上面的allocate函数以要创建的作业数量作为输入参数，生成最大值为 998的伪随机数，Job并使用该随机数和 for 循环计数器i作为 id 创建结构体，然后将其写入通道jobs。它在写入所有作业后关闭jobs通道。

下一步是创建读取results通道并打印输出的函数。

func result(done chan bool) {  
    for result := range results {
        fmt.Printf("Job id %d, input random no %d , sum of digits %d\n", result.job.id, result.job.randomno, result.sumofdigits)
    }
    done <- true
}
1
2
3
4
5
6

该result函数读取results通道并打印作业 ID、输入的随机号以及随机号的数字总和。结果函数还采用done通道作为参数，一旦打印完所有结果，就会写入该通道。

现在一切都准备好了。让我们继续完成从main()函数调用所有这些函数的最后一步。

func main() {  
    startTime := time.Now()
    noOfJobs := 100
    go allocate(noOfJobs)
    done := make(chan bool)
    go result(done)
    noOfWorkers := 10
    createWorkerPool(noOfWorkers)
    <-done
    endTime := time.Now()
    diff := endTime.Sub(startTime)
    fmt.Println("total time taken ", diff.Seconds(), "seconds")
}
1
2
3
4
5
6
7
8
9
10
11
12
13

我们首先将程序的执行开始时间存储在主函数的第 2 行中，在最后一行（第 12 行）中，我们计算 endTime 和 startTime 之间的时间差，并显示程序花费的总时间。这是必要的，因为我们将通过改变 Goroutine 的数量来进行一些基准测试。

设置noOfJobs为 100，然后调用allocate将作业添加到jobs通道。

然后创建done通道并将其传递给resultGoroutine，以便它可以开始打印输出并在所有内容打印完毕后发出通知。

最后，通过调用函数创建一个工作池，然后在done通道上等待打印所有结果。

这是完整的程序供您参考。我也导入了必要的包。

package main

import (  
    "fmt"
    "math/rand"
    "sync"
    "time"
)

type Job struct {  
    id       int
    randomno int
}
type Result struct {  
    job         Job
    sumofdigits int
}

var jobs = make(chan Job, 10)  
var results = make(chan Result, 10)

func digits(number int) int {  
    sum := 0
    no := number
    for no != 0 {
        digit := no % 10
        sum += digit
        no /= 10
    }
    time.Sleep(2 * time.Second)
    return sum
}
func worker(wg *sync.WaitGroup) {  
    for job := range jobs {
        output := Result{job, digits(job.randomno)}
        results <- output
    }
    wg.Done()
}
func createWorkerPool(noOfWorkers int) {  
    var wg sync.WaitGroup
    for i := 0; i < noOfWorkers; i++ {
        wg.Add(1)
        go worker(&wg)
    }
    wg.Wait()
    close(results)
}
func allocate(noOfJobs int) {  
    for i := 0; i < noOfJobs; i++ {
        randomno := rand.Intn(999)
        job := Job{i, randomno}
        jobs <- job
    }
    close(jobs)
}
func result(done chan bool) {  
    for result := range results {
        fmt.Printf("Job id %d, input random no %d , sum of digits %d\n", result.job.id, result.job.randomno, result.sumofdigits)
    }
    done <- true
}
func main() {  
    startTime := time.Now()
    noOfJobs := 100
    go allocate(noOfJobs)
    done := make(chan bool)
    go result(done)
    noOfWorkers := 10
    createWorkerPool(noOfWorkers)
    <-done
    endTime := time.Now()
    diff := endTime.Sub(startTime)
    fmt.Println("total time taken ", diff.Seconds(), "seconds")
}
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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

Run program in playground

请在本地计算机上运行此程序，以便更准确地计算总时间。

该程序将打印，

Job id 0, input random no 655 , sum of digits 16
Job id 4, input random no 150 , sum of digits 6 
Job id 1, input random no 146 , sum of digits 11
...
total time taken  20.079027 seconds
1
2
3
4
5

总共将打印 100 行，对应 100 个作业，最后一行将打印程序运行的总时间。您的输出将与我的不同，因为 Goroutines 可以按任何顺序运行，并且总时间也会根据硬件而有所不同。就我而言，程序大约需要 20 秒才能完成。

现在让我们将noOfWorkers中的增加到20。我们将工人数量增加了一倍。由于工作 Goroutine 增加了（准确地说增加了一倍），程序完成所需的总时间应该减少（准确地说减少一半）。就我而言，它变成了 10.035679 秒，程序打印出来，

Job id 19, input random no 915 , sum of digits 15
Job id 2, input random no 181 , sum of digits 10 
Job id 0, input random no 976 , sum of digits 22 
...
total time taken  10.035679 seconds  
1
2
3
4
5

现在我们可以理解，随着Goroutine 工作人数量的增加，完成作业所需的总时间会减少。我将其作为练习，让您使用不同值的函数中的noOfJobs和并分析结果。

本教程到此结束。祝你有美好的一天。