目前编程网上已经有很多关于Golang的文章了，自己在初次阅读这些文章中，也见识到了很多学习思路；那么本文《Worker Pool并发模式中何时关闭结果通道？》，也希望能帮助到大家，如果阅读完后真的对你学习Golang有帮助，欢迎动动手指，评论留言并分享~
问题内容我试图了解 golang 中的并发性是如何工作的，因为我正在尝试批量处理一个大型 json 文件，但我不确定何时应该关闭我的通道。下面是代码，
package main
import (
"fmt"
"os"
"time"
)
type order struct {
recipe string `json:"recipe"`
postcode string `json:"postcode"`
delivery timewithformat `json:"delivery"`
}
type timewithformat struct {
weekday string
starttime time.time
endtime time.time
valid bool
}
func main() {
input, err := os.open("test.json")
if err != nil {
fmt.println(err)
}
concurrency := 20
ingest := make(chan []order)
ret := make(chan map[string]int)
for x := 0; x < concurrency; x++ {
go processrecipesworker(ingest, ret)
}
go parseinmemory(input, ingest)
finalmap := make(map[string]int)
for v := range ret {
// combine all incoming results
for recipe, occurrence := range v {
if v, ok := finalmap[recipe]; ok {
finalmap[recipe] = occurrence + v
} else {
finalmap[recipe] = occurrence
}
}
}
//further processing on finalmap
}
func parseinmemory(input *os.file, ingest chan []order) {
data, _ := ioutil.readall(input)
var orders []order
if err := json.unmarshal(data, &orders); err != nil {
log.fatal(err.error())
}
batch := 100000
for i := 0; i < len(orders); i += batch {
j := i + batch
if j > len(orders) {
j = len(orders)
}
ingest <- orders[i:j]
}
close(ingest)
}
func processrecipesworker(in <-chan []helpers.order, r chan map[string]int) {
result := make(map[string]int)
for _, order := range <-in {
if v, ok := result[order.recipe]; ok {
result[order.recipe] = v + 1
} else {
result[order.recipe] = 1
}
}
r <- result
}我是 golang 和并发的新手，如果我的代码需要更改，请告诉我。
如有困惑，请随时发表评论。
编辑 torek 代码的建议如下所示，
package main
import (
"fmt"
"os"
"time"
)
func main() {
input, err := os.Open("test.json")
if err != nil {
fmt.Println(err)
}
var wg sync.WaitGroup
concurrency := 20
inGest := make(chan []helpers.Order)
ret := make(chan map[string]int)
wg.Add(concurrency)
for x := 0; x < concurrency; x++ {
go processRecipesWorker(inGest, ret)
}
go ParseInMemory(input, inGest)
go (func(wg *sync.WaitGroup) {
wg.Wait()
close(ret)
})(&wg)
finalMap := make(map[string]int)
for v := range ret {
// Combine all incoming results
for recipe, occurrence := range v {
if v, ok := finalMap[recipe]; ok {
finalMap[recipe] = occurrence + v
} else {
finalMap[recipe] = occurrence
}
}
}
//Further processing on finalMap
}
func ParseInMemory(input *os.File, ingest chan []Order) {
data, _ := ioutil.ReadAll(input)
var orders []Order
if err := json.Unmarshal(data, &orders); err != nil {
log.Fatal(err.Error())
}
batch := 100000
for i := 0; i < len(orders); i += batch {
j := i + batch
if j > len(orders) {
j = len(orders)
}
ingest <- orders[i:j]
}
close(ingest)
}
func processRecipesWorker(in <-chan []helpers.Order, r chan map[string]int) {
// This will get us unique recipes and there occurrences
// FR- 1 & 2 completed
result := make(map[string]int)
for _, order := range <-in {
if v, ok := result[order.Recipe]; ok {
result[order.Recipe] = v + 1
} else {
result[order.Recipe] = 1
}
}
r <- result
defer wg.Done()
}现在代码不会引发死锁，但当我有 10m 记录时它只处理 2m 记录，我知道它正在发生，因为所有例程都调用 done() 所以它结束了。但现在的问题是我什么时候应该调用 wg.done() 因为代码应该如何知道最后一批正在处理，我们可以调用 .done() 。
解决方案
你现在已经很接近了：
for _, order := range <-in {
...
}是错误的。这会接收一批并循环遍历它，但您希望循环遍历所有接收到的批次，然后循环遍历每个批次：
for batch := range in {
for _, order := range batch {
...
}
}在您的特定示例中，您知道要启动的 goroutine 数量，您可以在不使用 waitgroup 方法的情况下执行此操作。
Here 您可以找到代码修改的示例，为了简洁起见，我省略了您从文件中读取的部分。
它归结为两个主要修改。首先，您需要一个额外的通道（在我的示例中，它称为 quit）来跟踪每个 goroutine 何时完成工作。另外，还有已完成的 goroutine 的计数器（示例中为 donejobs）。
因此，您的 processrecipesworker 方法应如下所示：
func processrecipesworker(in <-chan []order, r chan map[string]int, quit chan<- int) {
result := make(map[string]int)
for orders := range in {
for _, order := range orders {
if v, ok := result[order.recipe]; ok {
result[order.recipe] = v + 1
} else {
result[order.recipe] = 1
}
}
}
r <- result
quit <- 1
}其次，您可以使用 for-select 循环，并带有标签来遍历发送到 ret 和 quit 通道的数据。
loop:
for{
select {
case v := <- ret:
// Combine all incoming results
for recipe, occurrence := range v {
if v, ok := finalMap[recipe]; ok {
finalMap[recipe] = occurrence + v
} else {
finalMap[recipe] = occurrence
}
}
case <-quit:
doneJobs++
fmt.Println(doneJobs)
if doneJobs >= concurrency {
break loop
}
}
}通过使用 break loop 语句，您可以中断 select 和 for 循环。
总而言之，@torek 使用 waitgroup 的解决方案是一个很好的解决方案，它提供了 go 中使用的通用模式的用法。您可以在 this 关于 go 管道的文章中看到这一点，特别是在扇出扇入部分。这个答案只是提供了另一种运行程序的方法，而不使用 waitgroup 方法。
今天关于《Worker Pool并发模式中何时关闭结果通道？》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于的内容请关注编程网公众号！
