我正在尝试使用Go中的并发性和渠道。我面临的问题主要是并发性，因此，我并没有放弃以下逻辑错误或应更改的逻辑。我有一个缓冲通道，其缓冲大小为“ N”，它还表示将要创建的goroutine的数量。所有例程都从一个通道读取并在另一个通道中写入，而主要的goroutine将从最后一个通道中打印值。1个输入通道--- N个goroutines查找并添加到输入和输出--- 1个输出通道问题是我总是陷入僵局，因为我不知道如何关闭正在馈送自身的通道，也不知道何时停止，因此我也无法关闭输出通道。该代码是以下示例：package mainconst count = 3const 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 channelfunc 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 channelfunc 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的所有数字并完成。谢谢
2 回答
牧羊人nacy
TA贡献1862条经验 获得超7个赞
我通常会发现，如果您在使设计模式或想法可行时遇到真正的问题，那说明您做错了。在这种情况下，自给式例行程序的想法知道何时应关闭自身。
我认为您正在寻找的是的想法worker pool。
本质上，您有一个包含的集合的通道，work然后是workersgo例程形式的多个例程，它们从该通道读取作业并对其进行处理，直到完成所有工作为止。
在以下示例中，我使用包gopool运行3个并发工作程序，这些工作程序由第4个go例程提供。
我等待所有工人关闭自己，这是由于输入工作通道被关闭而造成的。
// 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
- 2 回答
- 0 关注
- 190 浏览
添加回答
举报
0/150
提交
取消