Worker Pool并发模式中何时关闭结果通道？

目前编程网上已经有很多关于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并发模式中何时关闭结果通道？》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于的内容请关注编程网公众号！