Golang：新手——Master-Worker并发

Question

我在尝试实现这一点时遇到了问题（所有 goroutines 都睡着了 - 死锁！） 这是代码的要点：

var workers = runtime.NumCPU()

func main() {
    jobs := make(chan *myStruct, workers)
    done := make(chan *myStruct, workers)

    go produceWork(file_with_jobs, jobs)
    for i := 0; i < runtime.NumCPU(); i++ {
        go Worker(jobs, done)
    }
    consumeWork(done)
}

func produceWork(vf string, jobs chan *utils.DigSigEntries) {
    defer close(jobs)

    // load file with jobs
    file, err := ini.LoadFile(vf)

    // get data for processing
    for data, _ := range file {
        // ...
        jobs <- &myStruct{data1, data2, data3, false}
    }
}

func Worker(in, out chan *myStruct) {
    for {
        item, open := <-in
        if !open {
            break
        }

        process(item)
        out <- item
    }
    // close(out)   --> tried closing the out channel, but then not all items are processed
    //                  though no panics occur.
}

func process(item *myStruct) {
    //...modify the item
    item.status = true
}

func consumeWork(done chan *myStruct) {
    for val := range done {
        if !val.status {
            fmt.Println(val)
        }
    }
}

我主要是想了解如何在不使用同步/等待的东西的情况下做到这一点 - 只是纯频道 - 这可能吗？此例程的目标是让单个生产者加载由 N 个工人处理的项目 - 感谢任何指针/帮助。

Answer 1

您可以按照 siritinga 的建议，使用第三个信号或计数器通道，例如signal chan boolean，其中produceWork goroutine 会在每个作业进入jobs 通道之前添加一个值。因此，与jobs 一样，将向signal 传递相同数量的值：

func produceWork(vf string, jobs chan *utils.DigSigEntries, signal chan boolean) {
    defer close(jobs)

    // load file with jobs
    file, err := ini.LoadFile(vf)

    // get data for processing
    for data, _ := range file {
        // ...
        signal <- true
        jobs <- &myStruct{data1, data2, data3, false}
    }

    close(signal)
}

消费将从signal 频道开始读取。如果有一个值，则可以肯定会有一个从out 通道读取的值（一旦工作人员将其传递）。如果signal 关闭，那么一切都完成了。我们可以关闭剩余的done频道：

func consumeWork(done chan *myStruct, signal chan boolean) {
    for _ := range signal {
        val <- done
        if !val.status {
            fmt.Println(val)
        }
    }

    close(done)
} 

虽然这是可能的，但我不会真的推荐它。它不会使代码比使用sync.WaitGroup 时更清晰。毕竟，signal 频道基本上只能用作计数器。 WaitGroup 具有相同的目的，而且成本更低。

但您的问题不是关于如何解决问题，而是是否可以通过纯渠道解决问题。

Answer 2

抱歉，我没有注意到您想跳过 /sync :/ 我会留下答案，也许有人正在寻找这个。

import (
    "sync"
)


func main() {
    jobs := make(chan *myStruct, workers)
    done := make(chan *myStruct, workers)

    var workerWg sync.WaitGroup   // waitGroup for workers
    var consumeWg sync.WaitGroup  // waitGroup for consumer
    consumeWg.Add(1) // add one active Consumer

    for i := 0; i < runtime.NumCPU(); i++ {
        go Worker(&workerWg, jobs, done)
        workerWg.Add(1)
    }
    go consumeWork(&consumeWg, done)

    produceWork(file_with_jobs, jobs)

    close(jobs)
    workerWg.Wait()

    close(done)
    consumeWg.Wait()

}

func produceWork(vf string, jobs chan *utils.DigSigEntries) {

    // load file with jobs
    file, err := ini.LoadFile(vf)

    // get data for processing
    for data, _ := range file {
        // ...
        jobs <- &myStruct{data1, data2, data3, false}
    }
}


func Worker(wg *sync.WaitGroup, done chan *myStruct) {

    defer wg.Done()
    for job := range jobs {

        result := process(job)
        out <- result
    }

   // close(out)   --> tried closing the out channel, but then not all items are processed
   //                  though no panics occur.
}

func process(item *myStruct) {
    //...modify the item
    item.status = true
}

func consumeWork(wg *sync.WaitGroup, done chan *myStruct) {
    defer wg.Done()
    for val := range done {
        if !val.status {
            fmt.Println(val)
        }
    }
}