问题描述
我正在尝试在Go中创建简单的工作程序池。 在将等待组添加到以下程序后,我面临死锁。 其背后的核心原因是什么?
当我不使用等待组时,程序似乎运行正常。
Fatal error: all goroutines are asleep - deadlock!
goroutine 1 [semacquire]:
sync.runtime_Semacquire(0xc0001b2ea8)
程序-
package main
import (
"fmt"
"strconv"
"sync"
)
func main() {
workerSize := 2
ProcessData(workerSize)
}
// ProcessData :
func ProcessData(worker int) {
// Create Jobs Pool for passong jobs to worker
JobChan := make(chan string)
//Produce the jobs
var jobsArr []string
for i := 1; i <= 10000; i++ {
jobsArr = append(jobsArr,"Test "+strconv.Itoa(i))
}
//Assign jobs to worker from jobs pool
var wg sync.WaitGroup
for w := 1; w <= worker; w++ {
wg.Add(1)
// Consumer
go func(jw int,wg1 *sync.WaitGroup) {
defer wg1.Done()
for job := range JobChan {
actualProcess(job,jw)
}
}(w,&wg)
}
// Create jobs pool
for _,job := range jobsArr {
JobChan <- job
}
wg.Wait()
//close(JobChan)
}
func actualProcess(job string,worker int) {
fmt.Println("WorkerID: #",worker,",Job Value: ",job)
}
解决方法
一旦所有工作都消耗完了,您的工作人员将在for job := range JobChan
中等待更多数据。直到关闭通道,循环才会结束。
另一方面,您的主要goroutine正在等待wg.Wait()
且未达到(注释掉)关闭状态。
这时,所有goroutine都停留在等待数据或等待组完成的状态。
最简单的解决方案是在将所有作业发送到通道后直接调用close(JobChan)
:
// Create jobs pool
for _,job := range jobsArr {
JobChan <- job
}
close(JobChan)
wg.Wait()
,
对此进行了稍微的修改,但是是实现的更高级版本。我已经很好地注释了代码,以便于理解。因此,现在您可以配置作业数和作品数。甚至了解工作是如何在工人之间分配的,以便平均工作量几乎相等。
package main
import (
"fmt"
)
func main() {
var jobsCount = 10000 // Number of jobs
var workerCount = 2 // Number of workers
processData(workerCount,jobsCount)
}
func processData(workers,numJobs int) {
var jobsArr = make([]string,numJobs)
// jobArr with nTotal jobs
for i := 0; i < numJobs; i++ {
// Fill in jobs
jobsArr = append(jobsArr,fmt.Sprintf("Test %d",i+1))
}
var jobChan = make(chan string,1)
defer close(jobChan)
var (
// Length of jobsArr
length = len(jobsArr)
// Calculate average chunk size
chunks = len(jobsArr) / workers
// Window Start Index
wStart = 0
// Window End Index
wEnd = chunks
)
// Split the job between workers. Every workers gets a chunk of jobArr
// to work on. Distribution is work is approximately equal because last
// worker can less or more work as well.
for i := 1; i <= workers; i++ {
// Spawn a goroutine for every worker for chunk i.e.,jobArr[wStart:wEnd]
go func(wrk,s,e int) {
for j := s; j < e; j++ {
// Do some actual work. Send the actualProcess's return value to
// jobChan
jobChan <- actualProcess(wrk,jobsArr[j])
}
}(i,wStart,wEnd)
// Change pointers to get the set of chunk in next iteration
wStart = wEnd
wEnd += chunks
if i == workers-1 {
// If next worker is the last worker,// do till the end
wEnd = length
}
}
for i := 0; i < numJobs; i++ {
// Receieve all jobs
fmt.Println(<-jobChan)
}
}
func actualProcess(worker int,job string) string {
return fmt.Sprintf("WorkerID: #%d,Job Value: %s",worker,job)
}