使用Context、WaitGroup优雅处理Goroutine

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

你好，我是 Seekload。

今天给大家分享一篇 如何使用 context、waitGroup 实现程序快速且优雅退出 的文章!

原文如下：

最近，我正在编写一个“滴答器”的应用程序，每次“滴答”时可能会产生数千的 goroutine。我想确保当应用终止时，即使有一些特定的 goroutine 处理比较缓慢，它也能快速而优雅地退出。

刚开始的时候，围绕如何输出日志，我使用 sync.WaitGroup 实现流程控制，但我很快意识到如果我创建了很多 goroutine，即使其中很小一部分没有立即返回，我的程序会在终止时 hang 住。这让我重新考虑 context.WithCancel，并理解该如何重新调整我的程序，使其能快速且优雅地退出!

我们可以通过构建示例程序一步步来验证下，最初的示例程序并不会使用前面提到的技术点。

 
 
 
 
    
  
  
  
  
  1. package main 
    2. 
  
  
  
  
  2.  
    3. 
  
  
  
  
  3. import ( 
    4. 
  
  
  
  
  4.  "fmt" 
    5. 
  
  
  
  
  5.  "log" 
    6. 
  
  
  
  
  6.  "math/rand" 
    7. 
  
  
  
  
  7.  "os" 
    8. 
  
  
  
  
  8.  "os/signal" 
    9. 
  
  
  
  
  9.  "syscall" 
    10. 
  
  
  
  
  10.  "time" 
    11. 
  
  
  
  
  11. ) 
    12. 
  
  
  
  
  12.  
    13. 
  
  
  
  
  13. func doSomething(ch chan int) { 
    14. 
  
  
  
  
  14.  fmt.Printf("Received job %d\n", <-ch) 
    15. 
  
  
  
  
  15. } 
    16. 
  
  
  
  
  16.  
    17. 
  
  
  
  
  17. func init() { 
    18. 
  
  
  
  
  18.  rand.Seed(time.Now().Unix()) 
    19. 
  
  
  
  
  19. } 
    20. 
  
  
  
  
  20.  
    21. 
  
  
  
  
  21. func main() { 
    22. 
  
  
  
  
  22.  var ( 
    23. 
  
  
  
  
  23.   closing   = make(chan struct{}) 
    24. 
  
  
  
  
  24.   ticker    = time.NewTicker(1 * time.Second) 
    25. 
  
  
  
  
  25.   logger    = log.New(os.Stderr, "", log.LstdFlags) 
    26. 
  
  
  
  
  26.   batchSize = 6 
    27. 
  
  
  
  
  27.   jobs      = make(chan int, batchSize) 
    28. 
  
  
  
  
  28.  ) 
    29. 
  
  
  
  
  29.  
    30. 
  
  
  
  
  30.  go func() { 
    31. 
  
  
  
  
  31.   signals := make(chan os.Signal, 1) 
    32. 
  
  
  
  
  32.   signal.Notify(signals, syscall.SIGTERM, os.Interrupt) 
    33. 
  
  
  
  
  33.   <-signals 
    34. 
  
  
  
  
  34.   close(closing) 
    35. 
  
  
  
  
  35.  }() 
    36. 
  
  
  
  
  36. loop: 
    37. 
  
  
  
  
  37.  for { 
    38. 
  
  
  
  
  38.   select { 
    39. 
  
  
  
  
  39.   case <-closing: 
    40. 
  
  
  
  
  40.    break loop 
    41. 
  
  
  
  
  41.   case <-ticker.C: 
    42. 
  
  
  
  
  42.    for n := 0; n < batchSize; n++ { 
    43. 
  
  
  
  
  43.     jobs <- n 
    44. 
  
  
  
  
  44.     go doSomething(jobs) 
    45. 
  
  
  
  
  45.    } 
    46. 
  
  
  
  
  46.    logger.Printf("Completed doing %d things.", batchSize) 
    47. 
  
  
  
  
  47.   } 
    48. 
  
  
  
  
  48.  } 
    49. 
  
  
  
  
  49. } 
    50.

执行程序，我们会发现 Received job ... 和 Completed doing ... 会交替输出，输出可能类似下面这样：

 
 
 
 
    
  
  
  
  
  1. Received job 0 
    2. 
  
  
  
  
  2. Received job 1 
    3. 
  
  
  
  
  3. Received job 2 
    4. 
  
  
  
  
  4. 2021/02/08 21:30:59 Completed doing 6 things. 
    5. 
  
  
  
  
  5. Received job 3 
    6. 
  
  
  
  
  6. Received job 4 
    7. 
  
  
  
  
  7. Received job 5 
    8. 
  
  
  
  
  8. 2021/02/08 21:31:00 Completed doing 6 things. 
    9.

多次打印的结果并不一致!这是合理的，我们都知道 goroutines 并不会阻塞，所以除非我们对它做些什么，否则协程里的代码会立即执行。

我们添加 WaitGroup 来完善下流程，先在 var 代码块中定义变量：

 
 
 
 
    
  
  
  
  
  1. var ( 
    2. 
  
  
  
  
  2.     .. 
    3. 
  
  
  
  
  3.     wg sync.WaitGroup 
    4. 
  
  
  
  
  4. ) 
    5.

调整下 loop 循环：

 
 
 
 
    
  
  
  
  
  1. for n := 0; n < batchSize; n++ { 
    2. 
  
  
  
  
  2.     wg.Add(1) 
    3. 
  
  
  
  
  3.     jobs <- n 
    4. 
  
  
  
  
  4.     go doSomething(&wg, jobs) 
    5. 
  
  
  
  
  5. } 
    6. 
  
  
  
  
  6. wg.Wait() 
    7. 
  
  
  
  
  7. logger.Printf("Completed doing %d things.", batchSize) 
    8.

最后，修改协程函数：

 
 
 
 
    
  
  
  
  
  1. func doSomething(wg *sync.WaitGroup, ch chan int) { 
    2. 
  
  
  
  
  2.     defer wg.Done() 
    3. 
  
  
  
  
  3.     fmt.Printf("Received job %d\n", <-ch) 
    4. 
  
  
  
  
  4. } 
    5.

WaitGroups 会等待一组 goroutines 执行完成，仔细阅读代码我们发现：

  1. 每次循环时 WaitGroup 的计数器会加 1，加 1 原因是因为在 goroutine 里每次调用 wg.Done() 计数器会减一，这样 goroutine 执行完成返回之后计数器能维持平衡;
  2. 在调用 logger 之前，我们添加了 wg.Wait()，这样当程序执行到这里的时候便会阻塞直到 WaitGroups 的计数器减为 0。当所有 goroutines 调用 wg.Done() 之后，计数器便会恢复成 0。

很简单，是不是?我们再次执行程序，可以看到结果比之前的更一致：

 
 
 
 
    
  
  
  
  
  1. 2021/02/08 21:46:47 Completed doing 6 things. 
    2. 
  
  
  
  
  2. Received job 0 
    3. 
  
  
  
  
  3. Received job 1 
    4. 
  
  
  
  
  4. Received job 2 
    5. 
  
  
  
  
  5. Received job 4 
    6. 
  
  
  
  
  6. Received job 5 
    7. 
  
  
  
  
  7. Received job 3 
    8. 
  
  
  
  
  8. 2021/02/08 21:46:48 Completed doing 6 things. 
    9. 
  
  
  
  
  9. Received job 0 
    10. 
  
  
  
  
  10. Received job 2 
    11. 
  
  
  
  
  11. Received job 3 
    12. 
  
  
  
  
  12. Received job 4 
    13. 
  
  
  
  
  13. Received job 5 
    14. 
  
  
  
  
  14. Received job 1 
    15.

顺便说一句，与预期的一样，jobs 并不会按顺序执行，因为我们并没有采取任何措施来确保这一点。

在我们继续之前，按照目前的状态执行程序并尝试使用 Control+D 来终止程序，程序退出不会出现任何问题。

为了证明程序需要进一步完善，让我们添加一些代码模拟真实业务场景。我们新建一个函数，函数里面调用外部 API 并等待请求响应。请求过程中，我们将会调用 context.WithCancel 取消请求。

首先，创建一个未使用 context 的函数。下面的代码更复杂，有必要的话请看注释：

 
 
 
 
    
  
  
  
  
  1. func doAPICall(wg *sync.WaitGroup) error { 
    2. 
  
  
  
  
  2.  defer wg.Done() 
    3. 
  
  
  
  
  3.  
    4. 
  
  
  
  
  4.  req, err := http.NewRequest("GET", "https://httpstat.us/200", nil) 
    5. 
  
  
  
  
  5.  if err != nil { 
    6. 
  
  
  
  
  6.   return err 
    7. 
  
  
  
  
  7.  } 
    8. 
  
  
  
  
  8.  
    9. 
  
  
  
  
  9.  // The httpstat.us API accepts a sleep parameter which sleeps the request for the 
    10. 
  
  
  
  
  10.  // passed time in ms 
    11. 
  
  
  
  
  11.  q := req.URL.Query() 
    12. 
  
  
  
  
  12.  sleepMin := 1000 
    13. 
  
  
  
  
  13.  sleepMax := 4000 
    14. 
  
  
  
  
  14.  q.Set("sleep", fmt.Sprintf("%d", rand.Intn(sleepMax-sleepMin)+sleepMin)) 
    15. 
  
  
  
  
  15.  req.URL.RawQuery = q.Encode() 
    16. 
  
  
  
  
  16.  
    17. 
  
  
  
  
  17.  // Make the request to the API in an anonymous function, using a channel to 
    18. 
  
  
  
  
  18.  // communicate the results 
    19. 
  
  
  
  
  19.  c := make(chan error, 1) 
    20. 
  
  
  
  
  20.  go func() { 
    21. 
  
  
  
  
  21.   // For the purposes of this example, we're not doing anything with the response. 
    22. 
  
  
  
  
  22.   _, err := http.DefaultClient.Do(req) 
    23. 
  
  
  
  
  23.   c <- err 
    24. 
  
  
  
  
  24.  }() 
    25. 
  
  
  
  
  25.  
    26. 
  
  
  
  
  26.  // Block until the channel is populated 
    27. 
  
  
  
  
  27.  return <-c 
    28. 
  
  
  
  
  28. } 
    29.

修改定时器“滴答”，删除调用 doSomething() 的代码、删除 jobs channel(不会再使用到它)并且调用 doAPICall()。

 
 
 
 
    
  
  
  
  
  1. for n := 0; n < batchSize; n++ { 
    2. 
  
  
  
  
  2.     wg.Add(1) 
    3. 
  
  
  
  
  3.     go doAPICall(&wg) 
    4. 
  
  
  
  
  4. } 
    5.

执行程序并再次尝试退出程序：

  • WaitGroup 会等待所有的 goroutines 完成;
  • doAPICall() 调用会发生阻塞直到 httpstat.us() 接口返回，调用耗时大概 1000ms ~ 4000ms;
  • 取决于你终止程序的时间，退出会变得很困难(耗时比较长)，试一次可能发现不了问题，在不同的时刻多尝试几次;

现在来演示 context.WithCancel 如何进一步控制程序取消。当 context.WithCancel 初始化之后，会返回一个 context 和取消函数 CancelFunc()。这个取消函数会取消 context，第一次听到这个会困惑。阅读 Go 官方博客的文章 Go Concurrency Patterns: Context[1] 对于进一步理解 context.WithCancel 会有所帮助，推荐阅读完本篇文章之后再看!

ok，我们回到正文。为了实现取消流程控制，需要修改下代码。首先，使用 context 创建一个取消函数：

 
 
 
 
    
  
  
  
  
  1. var ( 
    2. 
  
  
  
  
  2.     ctx, cancel = context.WithCancel(context.Background()) 
    3. 
  
  
  
  
  3.     ... 
    4. 
  
  
  
  
  4. ) 
    5.

接着，在匿名函数里监听程序终止的信号，signals 被通知之后调用 CancelFunc，这意味着上下文将被视为已取消：

 
 
 
 
    
  
  
  
  
  1. go func() { 
    2. 
  
  
  
  
  2.     signals := make(chan os.Signal, 1) 
    3. 
  
  
  
  
  3.     signal.Notify(signals, syscall.SIGTERM, os.Interrupt) 
    4. 
  
  
  
  
  4.     <-signals 
    5. 
  
  
  
  
  5.     logger.Println("Initiating shutdown of producer.") 
    6. 
  
  
  
  
  6.     cancel() 
    7. 
  
  
  
  
  7.     close(closing) 
    8. 
  
  
  
  
  8. }() 
    9.

接着，调整 doAPICall() 函数，多接收一个 context 参数;使用 select-case 修改函数返回，等待 ctx.Done 或等待请求响应。为了简介，只展示了函数部分代码：

 
 
 
 
    
  
  
  
  
  1. func doAPICall(ctx context.Context, ....) { 
    2. 
  
  
  
  
  2.     // Cancel the request if ctx.Done is closed or await the response 
    3. 
  
  
  
  
  3.     select { 
    4. 
  
  
  
  
  4.     case <-ctx.Done(): 
    5. 
  
  
  
  
  5.            return ctx.Err() 
    6. 
  
  
  
  
  6.     case err := <-c: 
    7. 
  
  
  
  
  7.         return err 
    8. 
  
  
  
  
  8.     } 
    9. 
  
  
  
  
  9. } 
    10.

最后，确保调用 doAPICall() 函数时传递了 context 参数。现在，运行程序并多次在不同的时间点终止程序。

现在会发生什么?程序会立即退出。select-case 代码会监听 ctx.Done 是否关闭或者接口请求是否响应，哪个 case 的 channel 信号先到就先执行谁。当应用程序终止时，ctx.Done() 优先执行并且函数提前返回，不再关心请求是否响应。WaitGroup 的作用没变 - 等待一组 goroutines 完成。现在，程序的终止流程得到很大改善。

Go 的基本哲学之一就是：

Don't communicate by sharing memory; share memory by communicating.

这里，我们使用 channel 在 goroutines 之间传递引用，这使得我们能够改进应用程序的流程。

有很多种办法可以用来改善流程，例如，我们不跨 goroutine 接收 API 的响应或者错误。值得庆幸的是，Go 很容易就可以实现这点，因此可以将它视为一个起点，如果你还想完善，可以尝试下这些想法。

下面是完整的示例，仅供参考：

 
 
 
 
    
  
  
  
  
  1. package main 
    2. 
  
  
  
  
  2.  
    3. 
  
  
  
  
  3. import ( 
    4. 
  
  
  
  
  4.  "context" 
    5. 
  
  
  
  
  5.  "fmt" 
    6. 
  
  
  
  
  6.  "log" 
    7. 
  
  
  
  
  7.  "math/rand" 
    8. 
  
  
  
  
  8.  "net/http" 
    9. 
  
  
  
  
  9.  "os" 
    10. 
  
  
  
  
  10.  "os/signal" 
    11. 
  
  
  
  
  11.  "sync" 
    12. 
  
  
  
  
  12.  "syscall" 
    13. 
  
  
  
  
  13.  "time" 
    14. 
  
  
  
  
  14. ) 
    15. 
  
  
  
  
  15.  
    16. 
  
  
  
  
  16. func doAPICall(ctx context.Context, wg *sync.WaitGroup) error { 
    17. 
  
  
  
  
  17.  defer wg.Done() 
    18. 
  
  
  
  
  18.  
    19. 
  
  
  
  
  19.  req, err := http.NewRequest("GET", "https://httpstat.us/200", nil) 
    20. 
  
  
  
  
  20.  if err != nil { 
    21. 
  
  
  
  
  21.   return err 
    22. 
  
  
  
  
  22.  } 
    23. 
  
  
  
  
  23.  
    24. 
  
  
  
  
  24.  // The httpstat.us API accepts a sleep parameter which sleeps the request for the 
    25. 
  
  
  
  
  25.  // passed time in ms 
    26. 
  
  
  
  
  26.  q := req.URL.Query() 
    27. 
  
  
  
  
  27.  sleepMin := 1000 
    28. 
  
  
  
  
  28.  sleepMax := 4000 
    29. 
  
  
  
  
  29.  q.Set("sleep", fmt.Sprintf("%d", rand.Intn(sleepMax-sleepMin)+sleepMin)) 
    30. 
  
  
  
  
  30.  req.URL.RawQuery = q.Encode() 
    31. 
  
  
  
  
  31.  
    32. 
  
  
  
  
  32.  c := make(chan error, 1) 
    33. 
  
  
  
  
  33.  go func() { 
    34. 
  
  
  
  
  34.   // For the purposes of this example, we're not doing anything with the response. 
    35. 
  
  
  
  
  35.   _, err := http.DefaultClient.Do(req) 
    36. 
  
  
  
  
  36.   c <- err 
    37. 
  
  
  
  
  37.  }() 
    38. 
  
  
  
  
  38.  
    39. 
  
  
  
  
  39.  // Block until either channel is populated or closed 
    40. 
  
  
  
  
  40.  select { 
    41. 
  
  
  
  
  41.  case <-ctx.Done(): 
    42. 
  
  
  
  
  42.   return ctx.Err() 
    43. 
  
  
  
  
  43.  case err := <-c: 
    44. 
  
  
  
  
  44.   return err 
    45. 
  
  
  
  
  45.  } 
    46. 
  
  
  
  
  46. } 
    47. 
  
  
  
  
  47.  
    48. 
  
  
  
  
  48. func init() { 
    49. 
  
  
  
  
  49.  rand.Seed(time.Now().Unix()) 
    50. 
  
  
  
  
  50. } 
    51. 
  
  
  
  
  51.  
    52. 
  
  
  
  
  52. func main() { 
    53. 
  
  
  
  
  53.  var ( 
    54. 
  
  
  
  
  54.   closing     = make(chan struct{}) 
    55. 
  
  
  
  
  55.   ticker      = time.NewTicker(1 * time.Second) 
    56. 
  
  
  
  
  56.   logger      = log.New(os.Stderr, "", log.LstdFlags) 
    57. 
  
  
  
  
  57.   batchSize   = 6 
    58. 
  
  
  
  
  58.   wg          sync.WaitGroup 
    59. 
  
  
  
  
  59.   ctx, cancel = context.WithCancel(context.Background()) 
    60. 
  
  
  
  
  60.  ) 
    61. 
  
  
  
  
  61.  
    62. 
  
  
  
  
  62.  go func() { 
    63. 
  
  
  
  
  63.   signals := make(chan os.Signal, 1) 
    64. 
  
  
  
  
  64.   signal.Notify(signals, syscall.SIGTERM, os.Interrupt) 
    65. 
  
  
  
  
  65.   <-signals 
    66. 
  
  
  
  
  66.   cancel() 
    67. 
  
  
  
  
  67.   close(closing) 
    68. 
  
  
  
  
  68.  }() 
    69. 
  
  
  
  
  69. loop: 
    70. 
  
  
  
  
  70.  for { 
    71. 
  
  
  
  
  71.   select { 
    72. 
  
  
  
  
  72.   case <-closing: 
    73. 
  
  
  
  
  73.    break loop 
    74. 
  
  
  
  
  74.   case <-ticker.C: 
    75. 
  
  
  
  
  75.    for n := 0; n < batchSize; n++ { 
    76. 
  
  
  
  
  76.     wg.Add(1) 
    77. 
  
  
  
  
  77.     go doAPICall(ctx, &wg) 
    78. 
  
  
  
  
  78.    } 
    79. 
  
  
  
  
  79.    wg.Wait() 
    80. 
  
  
  
  
  80.    logger.Printf("Completed doing %d things.", batchSize) 
    81. 
  
  
  
  
  81.   } 
    82. 
  
  
  
  
  82.  } 
    83. 
  
  
  
  
  83. } 
    84.

最后一点，本文部分代码受到博文 Go Concurrency Patterns: Context[2] 的启发，再次推荐这篇文章。这篇文章还介绍了其他控制函数，比如：context.WithTimeout 等。Go 官方博客是每个人都应该阅读的宝库!

参考资料

[1]Go Concurrency Patterns: Context: https://blog.golang.org/context

[2]Go Concurrency Patterns: Context: https://blog.golang.org/context

via:https://justbartek.ca/p/golang-context-wg-go-routines/

作者：Bartek

网站栏目：使用Context、WaitGroup优雅处理Goroutine
网页网址：http://www.csdahua.cn/qtweb/news30/312480.html

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

广告

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

成都快上网为您推荐相关内容

网页设计公司知识

分类信息网站

重庆服务器托管　　 　四川成都展柜定制厂家　　 　内江做网站　　 　营销型网站建设　　 　成都纯水机　　 　域名注册　　 　微信小程序公众号开发　　 　简阳做网站　　 　大橙子建站　　 　网站建设　　 　绵阳电信服务器托管　　 　绵阳网站制作公司　　 　成都商城网站制作　　 　msfdjwx.com　　 　雅安主机托管　　 　外贸营销网站建设　　 　成都西信服务器托管　　 　全网整合营销　　 　贵安服务器托管　　 　模板商城网站