与渠道一起进入候车室工人

如何解决与渠道一起进入候车室工人

我正在尝试在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)
}