在 Go 中关闭自给自足通道

Question

我正在尝试在 Go 中使用并发和通道。我现在面临的问题主要是并发的思想，所以我不排斥下面的逻辑是错误的或者应该改的。

我有一个缓冲通道，它的缓冲区大小为“N”，它还表示将要创建的 goroutine 的数量。所有的例程都从一个通道读取并写入另一个通道，主 goroutine 将打印来自最终通道的值。

1 个输入通道 --- N 个 goroutine 查找并添加到输入和输出 --- 1 个输出通道

问题是我总是遇到死锁，因为我不知道如何关闭一个正在喂食的通道，也不知道它什么时候会停止，所以我也无法关闭输出通道。

代码如下：

package main

const count = 3
const finalNumber = 100

// There will be N routines running and reading from the one read channel
// The finalNumber is not known, in this examples is 100, but in the main problem will keep self feeding until the operation gives a wrong output
// readingRoutine will feed read channel and the print channel
func readingRoutine(read, print chan int) {
    for i := range read {
        print <- i
        if i < finalNumber && i+count < finalNumber {
            read <- i + count
        }
    }
}

// This is the main routine that will be printing the values from the print channel
func printingRoutine(print chan int) {
    for i := range print {
        println(i)
    }
}

func main() {
    read := make(chan int, count)
    print := make(chan int, count)

    // Feed count numbers into the buffered channel
    for i := 0; i < count; i++ {
        read <- i
    }

    // count go routines will be processing the read channel
    for i := 0; i < count; i++ {
        go readingRoutine(read, print)
    }
    printingRoutine(print)
}

在这个例子中，它应该打印从 0 到 100 的所有数字并完成。 谢谢

Answer 1

您可以使用sync.WaitGroup 等待事情完成，例如wg := &sync.WaitGroup{}

当您尝试打印finalNumber 次时，您应调用wg.Add(finalNumber)，然后在print() 中，每次打印完成后，调用wg.Done()。

生成另一个 goroutine 等待 wg.Wait() 然后关闭 read 通道和 print 通道。

func printingRoutine(print chan int,wg *sync.WaitGroup) {
    for i := range print {
        println(i)
        wg.Done()
    }
}

func main() {
    read := make(chan int, count)
    print := make(chan int, count)
    wg := &sync.WaitGroup{}
    wg.Add(finalNumber)

    // Feed count numbers into the buffered channel
    for i := 0; i < count; i++ {
        read <- i
    }

    // count go routines will be processing the read channel
    for i := 0; i < count; i++ {
        go readingRoutine(read, print)
    }
    go func() {
        wg.Wait()
        close(read)
        close(print)
    }()
    printingRoutine(print,wg)
}

游乐场：https://play.golang.org/p/BMSfz03egx0

Answer 2

我通常会发现，如果您在使设计模式或想法发挥作用时遇到真正的问题，那么您做错了。在这种情况下，自我喂食例程知道何时应该关闭自己的想法。

我认为您正在寻找的是worker pool 的想法。

基本上，您有一个频道，其中包含您的一组work，然后是一些workers，它们以 go 例程的形式从该频道读取作业并对其进行处理，直到所有工作完成。

在下面的示例中，我使用包 gopool 运行 3 个并发工作程序，这些工作程序由第 4 个 go 例程提供。

我等待所有worker都关闭自己，这是由于输入工作通道被关闭造成的。

// create the channel to store the jobs
// notice that we can only have 5 jobs in the channel at one time
workChan := make(chan int, 5)

// here we define what should be happening in each of our workers.
// each worker will be running concurrently
var work gopool.WorkFunc = func(ctx context.Context) error {
    for {
        select {
        case <-ctx.Done():
            // this is just a get out clause in case we want to prematurely stop the workers while there is still work to do
            return ctx.Err()
        case work, ok := <-workChan:
            if !ok {
                // we get here if the work channel has been closed
                return nil
            }
            // do something with work here
            fmt.Println(work)
        }
    }
}

// this func defines how many workers we want to be running concurrently
var workerCount gopool.WorkerCountFunc = func() uint64 {
    return 3
}

// here we define a new worker pool
p := gopool.NewPool("test", work, workerCount, nil, context.TODO())

// here we start the worker pool
cancel, err := p.StartOnce()
if err != nil {
        panic(err)
}

// the workers are now running and waiting for jobs

// we'll defer the cancel to make sure that the pool will be closed eventually
defer cancel()

// now we'll start a go routine to feed the workers
// it does this by adding to the workChan, and closes workChan when there is no more work to do
// when the workChan is closed the workers know that they should quit
go func(workChan chan<- int, min int, max int) {
    for i := min; i <= max; i++ {
        workChan <- i
    }
    close(workChan)
}(workChan, 3, 200)

// now we wait for the pool to be finished with it's work
<-p.Done()

fmt.Println("all work has been done")

示例输出：

$ go run main.go 
4
6
7
8
9
10
3
5
13
14
15
16
17
18
12
20
21
19
23
24
25
22
27
28
11
30
31
26
33
29
35
36
34
38
39
32
41
42
37
44
43
46
47
40
49
50
51
52
48
54
53
56
57
58
45
60
61
62
63
64
65
55
59
68
69
66
71
72
73
70
75
76
77
67
79
80
74
82
83
81
85
84
87
88
89
78
91
90
93
94
92
96
97
95
99
98
101
102
103
104
100
106
107
108
109
105
111
112
110
114
115
116
117
118
119
120
86
122
123
124
125
126
127
121
129
113
131
128
133
134
130
136
137
132
139
138
141
140
143
144
145
146
147
148
149
150
135
151
153
142
155
156
157
158
159
160
161
162
163
164
152
154
167
165
166
170
171
172
173
174
169
176
177
178
179
180
168
182
183
184
181
185
187
188
189
190
175
186
193
191
195
194
197
196
199
198
200
192
all work has been done