Go语言的并发与WorkerPool

本文转载自微信公众号「Golang来啦」，作者Seekload。转载本文请联系Golang来啦公众号。

创新互联-专业网站定制、快速模板网站建设、高性价比巨鹿网站开发、企业建站全套包干低至880元,成熟完善的模板库,直接使用。一站式巨鹿网站制作公司更省心,省钱,快速模板网站建设找我们，业务覆盖巨鹿地区。费用合理售后完善，十余年实体公司更值得信赖。

四哥水平有限，如有翻译或理解错误，烦请帮忙指出，感谢!

昨天分享关于 workerPool 的文章，有同学在后台说，昨天的 Demo 恰好符合项目的业务场景，真的非常棒!

所以今天就再来分享一篇 。

原文如下：

现代编程语言中，并发已经成为必不可少的特性。现在绝大多数编程语言都有一些方法实现并发。

其中一些实现方式非常强大，能将负载转移到不同的系统线程，比如 Java 等;一些则在同一线程上模拟这种行为，比如 Ruby 等。

Golang 的并发模型非常强大，称为 CSP(通信顺序进程)，它将一个问题分解成更小的顺序进程，然后调度这些进程的实例(称为 Goroutine)。这些进程通过 channel 传递信息实现通信。

本文，我们将探讨如何利用 golang 的并发性，以及如何在 workerPool 使用。系列文章的第二篇，我们将探讨如何构建一个强大的并发解决方案。

一个简单的例子

假设我们需要调用一个外部 API 接口，整个过程需要花费 100ms。如果我们需要同步地调用该接口 1000 次，则需要花费 100s。

 
 
 
    
  
  
  
  1. //// model/data.go
    2. 
  
  
  
  2. 
  
  
  
  3. package model
    4. 
  
  
  
  4. 
  
  
  
  5. type SimpleData struct {
    6. 
  
  
  
  6.  ID int
    7. 
  
  
  
  7. }
    8. 
  
  
  
  8. 
  
  
  
  9. //// basic/basic.go
    10. 
  
  
  
  10. 
  
  
  
  11. package basic
    12. 
  
  
  
  12. 
  
  
  
  13. import (
    14. 
  
  
  
  14.  "fmt"
    15. 
  
  
  
  15.  "github.com/Joker666/goworkerpool/model"
    16. 
  
  
  
  16.  "time"
    17. 
  
  
  
  17. )
    18. 
  
  
  
  18. 
  
  
  
  19. func Work(allData []model.SimpleData) {
    20. 
  
  
  
  20.  start := time.Now()
    21. 
  
  
  
  21.  for i, _ := range allData {
    22. 
  
  
  
  22.   Process(allData[i])
    23. 
  
  
  
  23.  }
    24. 
  
  
  
  24.  elapsed := time.Since(start)
    25. 
  
  
  
  25.  fmt.Printf("Took ===============> %s\n", elapsed)
    26. 
  
  
  
  26. }
    27. 
  
  
  
  27. 
  
  
  
  28. func Process(data model.SimpleData) {
    29. 
  
  
  
  29.  fmt.Printf("Start processing %d\n", data.ID)
    30. 
  
  
  
  30.  time.Sleep(100 * time.Millisecond)
    31. 
  
  
  
  31.  fmt.Printf("Finish processing %d\n", data.ID)
    32. 
  
  
  
  32. }
    33. 
  
  
  
  33. 
  
  
  
  34. //// main.go
    35. 
  
  
  
  35. 
  
  
  
  36. package main
    37. 
  
  
  
  37. 
  
  
  
  38. import (
    39. 
  
  
  
  39.  "fmt"
    40. 
  
  
  
  40.  "github.com/Joker666/goworkerpool/basic"
    41. 
  
  
  
  41.  "github.com/Joker666/goworkerpool/model"
    42. 
  
  
  
  42.  "github.com/Joker666/goworkerpool/worker"
    43. 
  
  
  
  43. )
    44. 
  
  
  
  44. 
  
  
  
  45. func main() {
    46. 
  
  
  
  46.  // Prepare the data
    47. 
  
  
  
  47.  var allData []model.SimpleData
    48. 
  
  
  
  48.  for i := 0; i < 1000; i++ {
    49. 
  
  
  
  49.   data := model.SimpleData{ ID: i }
    50. 
  
  
  
  50.   allData = append(allData, data)
    51. 
  
  
  
  51.  }
    52. 
  
  
  
  52.  fmt.Printf("Start processing all work \n")
    53. 
  
  
  
  53. 
  
  
  
  54.  // Process
    55. 
  
  
  
  55.  basic.Work(allData)
    56. 
  
  
  
  56. }
    57. 
 
 
 
 
 
 
    
  
  
  
  1. Start processing all work
    2. 
  
  
  
  2. Took ===============> 1m40.226679665s
    3.

上面的代码创建了 model 包，包里包含一个结构体，这个结构体只有一个 int 类型的成员。我们同步地处理 data，这显然不是最佳方案，因为可以并发处理这些任务。我们换一种方案，使用 goroutine 和 channel 来处理。

异步

 
 
 
    
  
  
  
  1. //// worker/notPooled.go
    2. 
  
  
  
  2. 
  
  
  
  3. func NotPooledWork(allData []model.SimpleData) {
    4. 
  
  
  
  4.  start := time.Now()
    5. 
  
  
  
  5.  var wg sync.WaitGroup
    6. 
  
  
  
  6. 
  
  
  
  7.  dataCh := make(chan model.SimpleData, 100)
    8. 
  
  
  
  8. 
  
  
  
  9.  wg.Add(1)
    10. 
  
  
  
  10.  go func() {
    11. 
  
  
  
  11.   defer wg.Done()
    12. 
  
  
  
  12.   for data := range dataCh {
    13. 
  
  
  
  13.    wg.Add(1)
    14. 
  
  
  
  14.    go func(data model.SimpleData) {
    15. 
  
  
  
  15.     defer wg.Done()
    16. 
  
  
  
  16.     basic.Process(data)
    17. 
  
  
  
  17.    }(data)
    18. 
  
  
  
  18.   }
    19. 
  
  
  
  19.  }()
    20. 
  
  
  
  20. 
  
  
  
  21.  for i, _ := range allData {
    22. 
  
  
  
  22.   dataCh <- allData[i]
    23. 
  
  
  
  23.  }
    24. 
  
  
  
  24. 
  
  
  
  25.  close(dataCh)
    26. 
  
  
  
  26.  wg.Wait()
    27. 
  
  
  
  27.  elapsed := time.Since(start)
    28. 
  
  
  
  28.  fmt.Printf("Took ===============> %s\n", elapsed)
    29. 
  
  
  
  29. }
    30. 
  
  
  
  30. 
  
  
  
  31. //// main.go
    32. 
  
  
  
  32. 
  
  
  
  33. // Process
    34. 
  
  
  
  34. worker.NotPooledWork(allData)
    35. 
 
 
 
 
 
 
    
  
  
  
  1. Start processing all work
    2. 
  
  
  
  2. Took ===============> 101.191534ms
    3.

上面的代码，我们创建了容量 100 的缓存 channel，并通过 NoPooledWork() 将数据 push 到 channel 里。channel 长度满 100 之后，我们是无法再向其中添加元素直到有元素被读取走。使用 for range 读取 channel，并生成 goroutine 处理。这里我们没有限制生成 goroutine 的数量，这可以尽可能多地处理任务。从理论上来讲，在给定所需资源的情况下，可以处理尽可能多的数据。执行代码，完成 1000 个任务只花费了 100ms。很疯狂吧!不全是，接着往下看。

问题

除非我们拥有地球上所有的资源，否则在特定时间内能够分配的资源是有限的。一个 goroutine 占用的最小内存是 2k，但也能达到 1G。上述并发执行所有任务的解决方案中，假设有一百万个任务，就会很快耗尽机器的内存和 CPU。我们要么升级机器的配置，要么就寻找其他更好的解决方案。

计算机科学家很久之前就考虑过这个问题，并提出了出色的解决方案 - 使用 Thread Pool 或者 Worker Pool。这个方案是使用 worker 数量受限的工作池来处理任务，workers 会按顺序一个接一个处理任务，这样就避免了 CPU 和内存使用急速增长。

解决方案：Worker Pool

我们通过实现 worker pool 来修复之前遇到的问题。

 
 
 
    
  
  
  
  1. //// worker/pooled.go
    2. 
  
  
  
  2. 
  
  
  
  3. func PooledWork(allData []model.SimpleData) {
    4. 
  
  
  
  4.  start := time.Now()
    5. 
  
  
  
  5.  var wg sync.WaitGroup
    6. 
  
  
  
  6.  workerPoolSize := 100
    7. 
  
  
  
  7. 
  
  
  
  8.  dataCh := make(chan model.SimpleData, workerPoolSize)
    9. 
  
  
  
  9. 
  
  
  
  10.  for i := 0; i < workerPoolSize; i++ {
    11. 
  
  
  
  11.   wg.Add(1)
    12. 
  
  
  
  12.   go func() {
    13. 
  
  
  
  13.    defer wg.Done()
    14. 
  
  
  
  14. 
  
  
  
  15.    for data := range dataCh {
    16. 
  
  
  
  16.     basic.Process(data)
    17. 
  
  
  
  17.    }
    18. 
  
  
  
  18.   }()
    19. 
  
  
  
  19.  }
    20. 
  
  
  
  20. 
  
  
  
  21.  for i, _ := range allData {
    22. 
  
  
  
  22.   dataCh <- allData[i]
    23. 
  
  
  
  23.  }
    24. 
  
  
  
  24. 
  
  
  
  25.  close(dataCh)
    26. 
  
  
  
  26.  wg.Wait()
    27. 
  
  
  
  27.  elapsed := time.Since(start)
    28. 
  
  
  
  28.  fmt.Printf("Took ===============> %s\n", elapsed)
    29. 
  
  
  
  29. }
    30. 
  
  
  
  30. 
  
  
  
  31. //// main.go
    32. 
  
  
  
  32. 
  
  
  
  33. // Process
    34. 
  
  
  
  34. worker.PooledWork(allData)
    35. 
 
 
 
 
 
 
    
  
  
  
  1. Start processing all work
    2. 
  
  
  
  2. Took ===============> 1.002972449s
    3.

上面的代码，worker 数量限制在 100，我们创建了相应数量的 goroutine 来处理任务。我们可以把 channel 看作是队列，worker goroutine 看作是消费者。多个 goroutine 可以监听同一个 channel，但是 channel 里的每一个元素只会被处理一次。

Go 语言的 channel 可以当作队列使用。

这是一个比较好的解决方案，执行代码，我们看到完成所有任务花费 1s。虽然没有 100ms 这么快，但已经能满足业务需要，而且我们得到了一个更好的解决方案，能将负载均摊在不同的时间片上。

处理错误

我们能做的还没完。上面看起来是一个完整的解决方案，但却不是的，我们没有处理错误情况。所以需要模拟出错的情形，并且看下我们需要怎么处理。

 
 
 
    
  
  
  
  1. //// worker/pooledError.go
    2. 
  
  
  
  2. 
  
  
  
  3. func PooledWorkError(allData []model.SimpleData) {
    4. 
  
  
  
  4.  start := time.Now()
    5. 
  
  
  
  5.  var wg sync.WaitGroup
    6. 
  
  
  
  6.  workerPoolSize := 100
    7. 
  
  
  
  7. 
  
  
  
  8.  dataCh := make(chan model.SimpleData, workerPoolSize)
    9. 
  
  
  
  9.  errors := make(chan error, 1000)
    10. 
  
  
  
  10. 
  
  
  
  11.  for i := 0; i < workerPoolSize; i++ {
    12. 
  
  
  
  12.   wg.Add(1)
    13. 
  
  
  
  13.   go func() {
    14. 
  
  
  
  14.    defer wg.Done()
    15. 
  
  
  
  15. 
  
  
  
  16.    for data := range dataCh {
    17. 
  
  
  
  17.     process(data, errors)
    18. 
  
  
  
  18.    }
    19. 
  
  
  
  19.   }()
    20. 
  
  
  
  20.  }
    21. 
  
  
  
  21. 
  
  
  
  22.  for i, _ := range allData {
    23. 
  
  
  
  23.   dataCh <- allData[i]
    24. 
  
  
  
  24.  }
    25. 
  
  
  
  25. 
  
  
  
  26.  close(dataCh)
    27. 
  
  
  
  27. 
  
  
  
  28.  wg.Add(1)
    29. 
  
  
  
  29.  go func() {
    30. 
  
  
  
  30.   defer wg.Done()
    31. 
  
  
  
  31.   for {
    32. 
  
  
  
  32.    select {
    33. 
  
  
  
  33.    case err := <-errors:
    34. 
  
  
  
  34.     fmt.Println("finished with error:", err.Error())
    35. 
  
  
  
  35.    case <-time.After(time.Second * 1):
    36. 
  
  
  
  36.     fmt.Println("Timeout: errors finished")
    37. 
  
  
  
  37.     return
    38. 
  
  
  
  38.    }
    39. 
  
  
  
  39.   }
    40. 
  
  
  
  40.  }()
    41. 
  
  
  
  41. 
  
  
  
  42.  defer close(errors)
    43. 
  
  
  
  43.  wg.Wait()
    44. 
  
  
  
  44.  elapsed := time.Since(start)
    45. 
  
  
  
  45.  fmt.Printf("Took ===============> %s\n", elapsed)
    46. 
  
  
  
  46. }
    47. 
  
  
  
  47. 
  
  
  
  48. func process(data model.SimpleData, errors chan<- error) {
    49. 
  
  
  
  49.  fmt.Printf("Start processing %d\n", data.ID)
    50. 
  
  
  
  50.  time.Sleep(100 * time.Millisecond)
    51. 
  
  
  
  51.  if data.ID % 29 == 0 {
    52. 
  
  
  
  52.   errors <- fmt.Errorf("error on job %v", data.ID)
    53. 
  
  
  
  53.  } else {
    54. 
  
  
  
  54.   fmt.Printf("Finish processing %d\n", data.ID)
    55. 
  
  
  
  55.  }
    56. 
  
  
  
  56. }
    57. 
  
  
  
  57. 
  
  
  
  58. //// main.go
    59. 
  
  
  
  59. 
  
  
  
  60. // Process
    61. 
  
  
  
  61. worker.PooledWorkError(allData)
    62.

我们修改了 process() 函数，处理一些随机的错误并将错误 push 到 errors chnanel 里。所以，为了处理并发出现的错误，我们可以使用 errors channel 保存错误数据。在所有任务处理完成之后，可以检查错误 channel 是否有数据。错误 channel 里的元素保存了任务 ID，方便需要的时候再处理这些任务。

比之前没处理错误，很明显这是一个更好的解决方案。但我们还可以做得更好，

我们将在下篇文章讨论如何编写一个强大的 worker pool 包，并且在 worker 数量受限的情况下处理并发任务。

总结

Go 语言的并发模型足够强大给力，只需要构建一个 worker pool 就能很好地解决问题而无需做太多工作，这就是它没有包含在标准库中的原因。但是，我们自己可以构建一个满足自身需求的方案。很快，我会在下一篇文章中讲到，敬请期待!

点击【阅读原文】直达代码仓库[1]。

参考资料

[1]代码仓库: https://github.com/Joker666/goworkerpool?ref=hackernoon.com

via:https://hackernoon.com/concurrency-in-golang-and-workerpool-part-1-e9n31ao

作者：Hasan

分享名称：Go语言的并发与WorkerPool
网页URL：http://www.shufengxianlan.com/qtweb/news13/524813.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

Google知识

分类信息网

尊格展览　　 　成都服务器租用　　 　商城网站建设公司　　 　网站建设　　 　成都光华机房服务器托管　　 　网站seo　　 　圣月阀门　　 　网络营销推广　　 　遂宁做网站　　 　卡多尼人工智　　 　广元双南建站　　 　成都400电话申请　　 　成都商城网站建设　　 　建站定制开发　　 　成都网站托管　　 　上市集团网站建设　　 　腾讯云香港虚拟主机　　 　平城网站运维　　 　南充护栏钻孔机　　 　商城系统开发