一个Demo学会WorkerPool

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

成都创新互联公司是工信部颁发资质IDC服务器商，为用户提供优质的服务器托管服务

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

今天给大家分享一篇关于 workPool 的文章，这个平时大家应该用的比较多，一起来看下。

原文如下：

工作池是这样一个池子，会创建指定数量的 worker，这些 worker 能获取任务并处理。允许多个任务同时处理，但是需要维持固定数量的 worker 避免系统资源被过度使用。

通常有两种方式创建任务池：

  • 一种是预先创建固定数量的 worker;
  • 另外一种是当有需要的时候才会创建 worker，当然也会有数量限制;

本文将与大家一起讨论第一种方式。当我们预先知道有许多任务需要同时运行，并且很大概率会用上最大数量的 worker，通常会采用这种方式。

为了演示，我们先创建 Worker 结构体，它获取任务并执行。

 
 
 
    
  
  
  
  1. import ( 
    2. 
  
  
  
  2.  "fmt" 
    3. 
  
  
  
  3. ) 
    4. 
  
  
  
  4.  
    5. 
  
  
  
  5. // Worker ... 
    6. 
  
  
  
  6. type Worker struct { 
    7. 
  
  
  
  7.  ID       int 
    8. 
  
  
  
  8.  Name     string 
    9. 
  
  
  
  9.  StopChan chan bool 
    10. 
  
  
  
  10. } 
    11. 
  
  
  
  11.  
    12. 
  
  
  
  12. // Start ... 
    13. 
  
  
  
  13. func (w *Worker) Start(jobQueue chan Job) { 
    14. 
  
  
  
  14.  w.StopChan = make(chan bool) 
    15. 
  
  
  
  15.  successChan := make(chan bool) 
    16. 
  
  
  
  16.  
    17. 
  
  
  
  17.  go func() { 
    18. 
  
  
  
  18.   successChan <- true 
    19. 
  
  
  
  19.   for { 
    20. 
  
  
  
  20.    // take job 
    21. 
  
  
  
  21.    job := <-jobQueue 
    22. 
  
  
  
  22.    if job != nil { 
    23. 
  
  
  
  23.     job.Start(w) 
    24. 
  
  
  
  24.    } else { 
    25. 
  
  
  
  25.     fmt.Printf("worker %s to be stopped\n", w.Name) 
    26. 
  
  
  
  26.     w.StopChan <- true 
    27. 
  
  
  
  27.     break 
    28. 
  
  
  
  28.    } 
    29. 
  
  
  
  29.   } 
    30. 
  
  
  
  30.  }() 
    31. 
  
  
  
  31.  
    32. 
  
  
  
  32.  // wait for the worker to start 
    33. 
  
  
  
  33.  <-successChan 
    34. 
  
  
  
  34. } 
    35. 
  
  
  
  35.  
    36. 
  
  
  
  36. // Stop ... 
    37. 
  
  
  
  37. func (w *Worker) Stop() { 
    38. 
  
  
  
  38.  // wait for the worker to stop, blocking 
    39. 
  
  
  
  39.  _ = <-w.StopChan 
    40. 
  
  
  
  40.  fmt.Printf("worker %s stopped\n", w.Name) 
    41. 
  
  
  
  41. } 
    42.

Worker 有一些属性保存当前的状态，另外还声明了两个方法分别用于启动、停止 worker。

在 Start() 方法里，创建了两个 channel 分别用于 worker 的启动和停止。最重要的是 for 循环里面，worker 会一直等待获取 job 并可执行的直到任务队列关闭。

Job 是包含单个方法 Start() 的接口，所以只要实现 Start() 方法就可以有不同类型的 job。

 
 
 
    
  
  
  
  1. // Job ... 
    2. 
  
  
  
  2. type Job interface { 
    3. 
  
  
  
  3.  Start(worker *Worker) error 
    4. 
  
  
  
  4. } 
    5.

一旦 Worker 确定之后，接下来就是创建 pool 来管理 workers。

 
 
 
    
  
  
  
  1. import ( 
    2. 
  
  
  
  2.  "fmt" 
    3. 
  
  
  
  3.  "sync" 
    4. 
  
  
  
  4. ) 
    5. 
  
  
  
  5.  
    6. 
  
  
  
  6. // Pool ... 
    7. 
  
  
  
  7. type Pool struct { 
    8. 
  
  
  
  8.  Name string 
    9. 
  
  
  
  9.  
    10. 
  
  
  
  10.  Size    int 
    11. 
  
  
  
  11.  Workers []*Worker 
    12. 
  
  
  
  12.  
    13. 
  
  
  
  13.  QueueSize int 
    14. 
  
  
  
  14.  Queue     chan Job 
    15. 
  
  
  
  15. } 
    16. 
  
  
  
  16.  
    17. 
  
  
  
  17. // Initiualize ... 
    18. 
  
  
  
  18. func (p *Pool) Initialize() { 
    19. 
  
  
  
  19.  // maintain minimum 1 worker 
    20. 
  
  
  
  20.  if p.Size < 1 { 
    21. 
  
  
  
  21.   p.Size = 1 
    22. 
  
  
  
  22.  } 
    23. 
  
  
  
  23.  p.Workers = []*Worker{} 
    24. 
  
  
  
  24.  for i := 1; i <= p.Size; i++ { 
    25. 
  
  
  
  25.   worker := &Worker{ 
    26. 
  
  
  
  26.    ID:   i - 1, 
    27. 
  
  
  
  27.    Name: fmt.Sprintf("%s-worker-%d", p.Name, i-1), 
    28. 
  
  
  
  28.   } 
    29. 
  
  
  
  29.   p.Workers = append(p.Workers, worker) 
    30. 
  
  
  
  30.  } 
    31. 
  
  
  
  31.  
    32. 
  
  
  
  32.  // maintain min queue size as 1 
    33. 
  
  
  
  33.  if p.QueueSize < 1 { 
    34. 
  
  
  
  34.   p.QueueSize = 1 
    35. 
  
  
  
  35.  } 
    36. 
  
  
  
  36.  p.Queue = make(chan Job, p.QueueSize) 
    37. 
  
  
  
  37. } 
    38. 
  
  
  
  38.  
    39. 
  
  
  
  39. // Start ... 
    40. 
  
  
  
  40. func (p *Pool) Start() { 
    41. 
  
  
  
  41.  for _, worker := range p.Workers { 
    42. 
  
  
  
  42.   worker.Start(p.Queue) 
    43. 
  
  
  
  43.  } 
    44. 
  
  
  
  44.  fmt.Println("all workers started") 
    45. 
  
  
  
  45. } 
    46. 
  
  
  
  46.  
    47. 
  
  
  
  47. // Stop ... 
    48. 
  
  
  
  48. func (p *Pool) Stop() { 
    49. 
  
  
  
  49.  close(p.Queue) // close the queue channel 
    50. 
  
  
  
  50.  
    51. 
  
  
  
  51.  var wg sync.WaitGroup 
    52. 
  
  
  
  52.  for _, worker := range p.Workers { 
    53. 
  
  
  
  53.   wg.Add(1) 
    54. 
  
  
  
  54.   go func(w *Worker) { 
    55. 
  
  
  
  55.    defer wg.Done() 
    56. 
  
  
  
  56.  
    57. 
  
  
  
  57.    w.Stop() 
    58. 
  
  
  
  58.   }(worker) 
    59. 
  
  
  
  59.  } 
    60. 
  
  
  
  60.  wg.Wait() 
    61. 
  
  
  
  61.  fmt.Println("all workers stopped") 
    62. 
  
  
  
  62. } 
    63.

Pool 包含 worker 切片和用于保存 job 的队列。worker 的数量在初始化的时候是可以自定义。

关键点在 Stop() 的逻辑，当它被调用时，会先关闭 job 队列，worker 便会从 job 队列读到 nil，接着就会关闭对应的 worker。接着在 for 循环里，等待 worker 并发地停止直到最后一个 worker 停止。

为了演示整体逻辑，下面的例子展示了一个仅仅输出值的 job。

 
 
 
    
  
  
  
  1. import "fmt" 
    2. 
  
  
  
  2.  
    3. 
  
  
  
  3. func main() { 
    4. 
  
  
  
  4.  pool := &Pool{ 
    5. 
  
  
  
  5.   Name:      "test", 
    6. 
  
  
  
  6.   Size:      5, 
    7. 
  
  
  
  7.   QueueSize: 20, 
    8. 
  
  
  
  8.  } 
    9. 
  
  
  
  9.  pool.Initialize() 
    10. 
  
  
  
  10.  pool.Start() 
    11. 
  
  
  
  11.         defer pool.Stop() 
    12. 
  
  
  
  12.  
    13. 
  
  
  
  13.  for i := 1; i <= 100; i++ { 
    14. 
  
  
  
  14.   job := &PrintJob{ 
    15. 
  
  
  
  15.    Index: i, 
    16. 
  
  
  
  16.   } 
    17. 
  
  
  
  17.   pool.Queue <- job 
    18. 
  
  
  
  18.  } 
    19. 
  
  
  
  19. } 
    20. 
  
  
  
  20.  
    21. 
  
  
  
  21. // PrintJob ... 
    22. 
  
  
  
  22. type PrintJob struct { 
    23. 
  
  
  
  23.  Index int 
    24. 
  
  
  
  24. } 
    25. 
  
  
  
  25.  
    26. 
  
  
  
  26. func (pj *PrintJob) Start(worker *Worker) error { 
    27. 
  
  
  
  27.  fmt.Printf("job %s - %d\n", worker.Name, pj.Index) 
    28. 
  
  
  
  28.  return nil 
    29. 
  
  
  
  29. } 
    30.

如果你看了上面的代码逻辑，就会发现很简单，创建了有 5 个 worker 的工作池并且 job 队列的大小是 20。

接着，模拟 job 创建和处理过程：一旦 job 被创建就会 push 到任务队列里，等待着的 worker 便会从队列里取出 job 并处理。

类似下面这样的输出：

 
 
 
    
  
  
  
  1. all workers started 
    2. 
  
  
  
  2. job test-worker-3 - 4 
    3. 
  
  
  
  3. job test-worker-3 - 6 
    4. 
  
  
  
  4. job test-worker-3 - 7 
    5. 
  
  
  
  5. job test-worker-3 - 8 
    6. 
  
  
  
  6. job test-worker-3 - 9 
    7. 
  
  
  
  7. job test-worker-3 - 10 
    8. 
  
  
  
  8. job test-worker-3 - 11 
    9. 
  
  
  
  9. job test-worker-3 - 12 
    10. 
  
  
  
  10. job test-worker-3 - 13 
    11. 
  
  
  
  11. job test-worker-3 - 14 
    12. 
  
  
  
  12. job test-worker-3 - 15 
    13. 
  
  
  
  13. job test-worker-3 - 16 
    14. 
  
  
  
  14. job test-worker-3 - 17 
    15. 
  
  
  
  15. job test-worker-3 - 18 
    16. 
  
  
  
  16. job test-worker-3 - 19 
    17. 
  
  
  
  17. job test-worker-3 - 20 
    18. 
  
  
  
  18. worker test-worker-3 to be stopped 
    19. 
  
  
  
  19. job test-worker-4 - 5 
    20. 
  
  
  
  20. job test-worker-0 - 1 
    21. 
  
  
  
  21. worker test-worker-3 stopped 
    22. 
  
  
  
  22. job test-worker-2 - 3 
    23. 
  
  
  
  23. worker test-worker-2 to be stopped 
    24. 
  
  
  
  24. worker test-worker-2 stopped 
    25. 
  
  
  
  25. worker test-worker-4 to be stopped 
    26. 
  
  
  
  26. worker test-worker-4 stopped 
    27. 
  
  
  
  27. worker test-worker-0 to be stopped 
    28. 
  
  
  
  28. worker test-worker-0 stopped 
    29. 
  
  
  
  29. job test-worker-1 - 2 
    30. 
  
  
  
  30. worker test-worker-1 to be stopped 
    31. 
  
  
  
  31. worker test-worker-1 stopped 
    32. 
  
  
  
  32. all workers stopped 
    33.

via:https://www.pixelstech.net/article/1611483826-Demo-on-creating-worker-pool-in-GoLang

作者：sonic0002

当前文章：一个Demo学会WorkerPool
本文路径：http://www.gawzjz.com/qtweb2/news5/27505.html

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

广告

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

猜你还喜欢下面的内容

建站公司知识

行业网站建设

广东帝美豪门窗　　 　成都广告制作　　 　恒温恒湿空调机组　　 　成都机柜租用　　 　云主机　　 　娃娃鱼鱼苗　　 　重庆网站建设　　 　马边网站建设　　 　艾瑞格衣柜　　 　香港云服务器　　 　搜索引擎优化　　 　谷歌推广　　 　微信小程序开发　　 　云主机　　 　成都电商网站建设方案　　 　微信网站开发　　 　国泰民安消防　　 　成都全网营销推广　　 　正泰动物　　 　乐山企业网站设计