问题内容
我只想创建一定数量的 go 例程，比如 5 个，但我可以接收可变数量的作业。
这是我尝试执行此操作的代码，测试位于其下方。
package main
import (
"context"
"fmt"
"runtime"
"time"
)
func dowork(size int, capacity int) int {
start := time.now()
jobs := make(chan *job, capacity)
results := make(chan *job, capacity)
sem := make(chan struct{}, capacity)
go chanworker(jobs, results, sem)
for i := 0; i < size; i++ {
jobs <- &job{id: i}
}
close(jobs)
successcount := 0
for i := 0; i < size; i++ {
item := <-results
if item.result {
successcount++
}
fmt.printf("job %d completed %v\n", item.id, item.result)
}
close(results)
close(sem)
fmt.printf("time taken to execute %d jobs with %d capacity = %v\n", size, capacity, time.since(start))
return successcount
}
func chanworker(jobs <-chan *job, results chan<- *job, sem chan struct{}) {
for item := range jobs {
it := item
sem <- struct{}{}
fmt.printf("job %d started\n", it.id)
go func() {
timeoutctx, cancel := context.withtimeout(context.background(), 300*time.millisecond)
defer cancel()
time.sleep(time.duration(it.id) * 100 * time.millisecond)
select {
case <-timeoutctx.done():
fmt.printf("job %d timed out\n", it.id)
it.result = false
results <- it
<-sem
return
default:
fmt.printf("total number of routines %d\n", runtime.numgoroutine())
it.result = true
results <- it
<-sem
}
}()
}
}
对此的测试
package main
import (
"testing"
)
func Test_doWork(t *testing.T) {
type args struct {
size int
capacity int
}
tests := []struct {
name string
args args
want int
}{
{
name: "jobs 10 capacity 5",
args: args{
size: 10,
capacity: 5,
},
want: 3,
},
{
name: "jobs 100 capacity 5",
args: args{
size: 100,
capacity: 5,
},
want: 3,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if got := doWork(tt.args.size, tt.args.capacity); got < tt.want {
t.Errorf("doWork() = %v, want %v", got, tt.want)
}
})
}
}
测试 jobs 10 容量 5 有效，但 jobs 100 容量 5 失败。
如果我为 100 个作业设置容量 50，它可以工作，但对于 30 个作业则不起作用，并且无法理解其行为。
以下是我对渠道的理解并期望它能够发挥作用。
缓冲通道如果已满，将会阻塞，直到有一些空闲容量可用。我预计一旦 jobs 通道满了，它就会阻塞，直到 chanworker 释放其中的一些。 chanworker 本身接收一个容量并使用空结构来确保创建的工作线程不超过 5 个。
为什么我会遇到错误
致命错误：所有 goroutine 都在休眠 - 死锁！ ？
正确答案
由于主 goroutine 在所有作业都发送到 jobs 之前不会从 results 接收值，因此工作线程会在发送到 results 时阻塞。主 goroutine 阻止发送到 jobs 因为工作被阻止。陷入僵局！
通过使用 goroutine 来完成工作来修复。
go func() {
for i := 0; i < size; i++ {
jobs <- &Job{id: i}
}
close(jobs)
}()
