golang协程关闭踩坑实战记录
作者：文大侠666 发布时间：2024-02-13 00:02:57
前言
Go语言中，协程创建和启动非常简单，但是如何才能正确关闭协程呢，和开车一样，前进总是很容易，但是如何正确的把车停在指定的地方总是不容易的。生产实践中，go常常遇到未能正确关闭协程而影响程序运行的场景，轻则协程泄漏资源浪费，重则程序崩溃。
本文，总结协程关闭的三大原则，结合实际场景让你彻底搞定协程关闭，保证又快又稳！
场景
结合如下典型场景，主进程中起多个协程，这些协程会
1.共同消费一个数据通道 data channel
2.也可能共享一个退出通道channel或context
那么，应该如何正确关闭呢
原则1-协程接受通知主动关闭
并不推荐强制停止，更多的时候我们希望在停止时，干一点事比如资源清理/连接清理等，这时候最好的方式就是通知协程退出，具体何时退出和退出前做什么完全由当前要关闭的协程控制。
通知一般有三种方式
data channel关闭通知退出
适用简单任务，复杂的更推荐context单独通知
// cancelFn 数据通道关闭通知退出
func cancelFn(dataChan chan int) {
for {
select {
case val, ok := <-dataChan:
// 关闭data通道时，通知退出
// 一个可选是判断data=指定值时退出
if !ok {
log.Printf("Channel closed ！！！")
return
}
log.Printf("Revice dataChan ％d\n", val)
}
}
}
exit channel关闭通知退出
部分简单场景适用
// exitChannelFn 单独退出通道关闭通知退出
func exitChannelFn(wg *sync.WaitGroup, taskNo int, dataChan chan int, exitChan chan struct{}) {
defer wg.Done()
for {
select {
case val, ok := <-dataChan:
if !ok {
log.Printf("Task ％d channel closed ！！！", taskNo)
return
}
log.Printf("Task ％d revice dataChan ％d\n", taskNo, val)
// 关闭exit通道时，通知退出
case <-exitChan:
log.Printf("Task ％d revice exitChan signal!\n", taskNo)
return
}
}
}
context超时或取消通知退出
主流推荐
// contextCancelFn context取消或超时通知退出
func contextCancelFn(wg *sync.WaitGroup, taskNo int, dataChan chan int, ctx context.Context) {
defer wg.Done()
for {
select {
case val, ok := <-dataChan:
if !ok {
log.Printf("Task ％d channel closed ！！！", taskNo)
return
}
log.Printf("Task ％d revice dataChan ％d\n", taskNo, val)
// ctx取消或超时，通知退出
case <-ctx.Done():
log.Printf("Task ％d revice exit signal!\n", taskNo)
return
}
}
}
原则2-谁负责创建协程谁负责关闭协程
go func可以立即创建一个协程，因此常常遇到我们可能在任何一个地方创建协程，但是在哪里关闭呢，是需要统一管理吗？官方推荐的最佳实践就是，谁负责创建协程谁负责关闭协程。
参考如下，每次调用execDataTaskFunc函数执行都会起一个协程异步执行，协程关闭通过监控外层函数context参数来实现。
func execDataTaskFunc(ctx context.Context, dataChan chan int, taskName string) chan int {
out := make(chan int)
log.Printf("Task ％s start!\n", taskName)
go func() {
defer close(out)
for {
select {
case data, ok := <-dataChan:
if !ok {
log.Printf("Task ％s revice data channel close signal!\n", taskName)
return
}
// do something
out <- data
case <-ctx.Done():
log.Printf("Task ％s revice exit signal!\n", taskName)
return
}
}
}()
return out
}
原则3-等待所有协程关闭再退出
通常对于正在运行的协程，发出退出通知后，具体程序何时才能退出呢？一般如下三种方式
WaitGroup/ErrGroup判断所有协程关闭后退出
最常用，参考如下
// 多个任务并行控制，等待所有任务完成
func TestTaskControl(t *testing.T) {
dataChan := make(chan int)
taskNum := 3
wg := sync.WaitGroup{}
wg.Add(taskNum)
// 起多个协程，data关闭时退出
for i := 0; i < taskNum; i++ {
go func(taskNo int) {
defer wg.Done()
t.Logf("Task ％d run\n", taskNo)
for {
select {
case _, ok := <-dataChan:
if !ok {
t.Logf("Task ％d notify to stop\n", taskNo)
return
}
}
}
}(i)
}
// 通知退出
go func() {
time.Sleep(3 * time.Second)
close(dataChan)
}()
// 等待退出完成
wg.Wait()
}
等待channel关闭后退出
参考如下，对于部分任务场景，协程数据输出到新建的channel中，可以在此channel上阻塞等待，直到协程通知关闭时，关闭此channel然后程序退出。
// 多个任务并行控制，等待所有任务完成
func TestTaskControl2(t *testing.T) {
dataChan := make(chan int)
// 起协程返回新chan，在输出chan等待判断完成
out := make(chan int)
go func() {
defer close(out) // 结束则自动关闭
for {
select {
case _, ok := <-dataChan:
if !ok {
t.Logf("Task notify to stop\n")
return
}
}
}
}()
// 通知退出
go func() {
time.Sleep(3 * time.Second)
close(dataChan)
}()
dataChan <- 1
// 等待退出完成
for data := range out {
t.Logf("％d\n", data)
}
}
等待足够长时间后关闭
对于部分任务，能够估算从通知关闭到实际关闭时间，则可等待足够长时间来保证协程关闭然后退出，实际场景并不推荐，带有一定不确定性，很容易出错。
func TestTaskControl3(t *testing.T) {
dataChan := make(chan int)
// 起协程返回新chan
out := make(chan int)
go func() {
defer close(out) // 结束则自动关闭
for {
select {
case _, ok := <-dataChan:
if !ok {
t.Logf("Task notify to stop\n")
return
}
}
}
}()
// 通知退出
go func() {
time.Sleep(3 * time.Second)
close(dataChan)
}()
dataChan <- 1
// 等待足够长时间，退出完成
time.Sleep(10 * time.Second)
}
复杂退出场景
结合三大原则，这里展示部分复杂场景的协程关闭方案。
嵌套协程，同时关闭
如下是多个任务执行，每个任务一个协程，现在考虑如下目标
支持多级嵌套，父任务停止后，子任务自动停止
方案：使用context通知，WaitGroup等待所有任务关闭后退出
任务运行代码
type TaskFunc func(ctx context.Context)
func runTaskFunc(wg *sync.WaitGroup, ctx context.Context, taskName string, f TaskFunc) {
defer wg.Done()
log.Printf("Task ％s start!\n", taskName)
f(ctx)
for {
select {
case <-ctx.Done():
log.Printf("Task ％s revice exit signal!\n", taskName)
return
}
}
}
整体实现代码
// 简单并行任务-同时停止
func TestStop(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
var wg = sync.WaitGroup{}
// 起多个任务
wg.Add(1)
go runTaskFunc(&wg, ctx, "A", func(ctx context.Context) {
wg.Add(1)
go runTaskFunc(&wg, ctx, "B", func(ctx context.Context) {
wg.Add(1)
go runTaskFunc(&wg, ctx, "C", func(ctx context.Context) {
wg.Add(1)
go runTaskFunc(&wg, ctx, "D", func(ctx context.Context) {})
})
})
wg.Add(1)
go runTaskFunc(&wg, ctx, "E", func(ctx context.Context) {
wg.Add(1)
go runTaskFunc(&wg, ctx, "F", func(ctx context.Context) {
wg.Add(1)
go runTaskFunc(&wg, ctx, "G", func(ctx context.Context) {})
})
})
})
// 通知关闭
go func() {
time.Sleep(3 * time.Second)
cancel()
}()
// 等待全部关闭后退出
wg.Wait()
}
协程关闭是无序的，如下
2023/01/07 22:40:09 Task A start!
2023/01/07 22:40:09 Task E start!
2023/01/07 22:40:09 Task F start!
2023/01/07 22:40:09 Task G start!
2023/01/07 22:40:09 Task B start!
2023/01/07 22:40:09 Task C start!
2023/01/07 22:40:09 Task D start!
2023/01/07 22:40:12 Task A revice exit signal!
2023/01/07 22:40:12 Task G revice exit signal!
2023/01/07 22:40:12 Task B revice exit signal!
2023/01/07 22:40:12 Task F revice exit signal!
2023/01/07 22:40:12 Task D revice exit signal!
2023/01/07 22:40:12 Task C revice exit signal!
2023/01/07 22:40:12 Task E revice exit signal!
嵌套协程，指定顺序关闭
还是上述场景，现在需求是：控制停止顺序，先停EFG 再停BCD 最后停A
方案：借助context通知，指定多个cancel点，WaitGroup等待所有任务关闭后退出
// 简单并行任务-控制停止顺序
func TestStop2(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
ctxb, cancelb := context.WithCancel(ctx)
ctxe, cancele := context.WithCancel(ctx)
var wg = sync.WaitGroup{}
// 起多个任务
wg.Add(1)
go runTaskFunc(&wg, ctx, "A", func(ctx context.Context) {
wg.Add(1)
go runTaskFunc(&wg, ctxb, "B", func(ctx context.Context) {
wg.Add(1)
go runTaskFunc(&wg, ctx, "C", func(ctx context.Context) {
wg.Add(1)
go runTaskFunc(&wg, ctx, "D", func(ctx context.Context) {})
})
})
wg.Add(1)
go runTaskFunc(&wg, ctxe, "E", func(ctx context.Context) {
wg.Add(1)
go runTaskFunc(&wg, ctx, "F", func(ctx context.Context) {
wg.Add(1)
go runTaskFunc(&wg, ctx, "G", func(ctx context.Context) {})
})
})
})
// 通知关闭
go func() {
time.Sleep(3 * time.Second)
cancele()
time.Sleep(3 * time.Second)
cancelb()
time.Sleep(3 * time.Second)
cancel()
}()
// 等待全部关闭后退出
wg.Wait()
}
运行如下，协程按照指定顺序关闭
2023/01/07 22:40:40 Task A start!
2023/01/07 22:40:40 Task E start!
2023/01/07 22:40:40 Task F start!
2023/01/07 22:40:40 Task G start!
2023/01/07 22:40:40 Task B start!
2023/01/07 22:40:40 Task C start!
2023/01/07 22:40:40 Task D start!
2023/01/07 22:40:43 Task E revice exit signal!
2023/01/07 22:40:43 Task F revice exit signal!
2023/01/07 22:40:43 Task G revice exit signal!
2023/01/07 22:40:46 Task B revice exit signal!
2023/01/07 22:40:46 Task D revice exit signal!
2023/01/07 22:40:46 Task C revice exit signal!
2023/01/07 22:40:49 Task A revice exit signal!
嵌套协程，逐级关闭
考虑如下场景，A->B->C嵌套起协程，每个协程创建新的channel传输数据给下游
如下起任务，每个任务可以通过context或者data channel关闭来通知退出
func execDataTaskFunc(ctx context.Context, dataChan chan int, taskName string) chan int {
out := make(chan int)
//out := make(chan int, 100)
log.Printf("Task ％s start!\n", taskName)
go func() {
defer close(out)
for {
select {
case data, ok := <-dataChan:
if !ok {
log.Printf("Task ％s revice data channel close signal!\n", taskName)
return
}
time.Sleep(2 * time.Second)
out <- data
case <-ctx.Done():
log.Printf("Task ％s revice exit signal!\n", taskName)
return
}
}
}()
return out
}
整体流程如下
func TestDataTaskStop(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
dataChanInput := make(chan int)
// 嵌套运行协程
taskChanA := execDataTaskFunc(ctx, dataChanInput, "A")
taskChanB := execDataTaskFunc(ctx, taskChanA, "B")
taskChanC := execDataTaskFunc(ctx, taskChanB, "C")
// 通知退出
go func() {
i := 0
for {
select {
case <-time.After(time.Second):
i = i + 1
if i == 10 {
t.Logf("Notify to stop!!!")
close(dataChanInput)
//cancel()
return
}
dataChanInput <- i
}
}
}()
// 等待退出
for data := range taskChanC {
t.Logf("Out->％d", data)
}
}
这里数据每条数据产生间隔1秒，每个任务处理时长为2秒，也就是说通知关闭时，可能上游任务处理中，下游还没来得及处理，因此期望的是逐级依次关闭A/B/C，确保上游数据处理完成传给下游，不要丢失数据。
对比context通知退出和data channel关闭通知退出，对比如下。可以看到如果我们是有中间处理和逐级关闭需求的还是要依赖close关闭协程来通知，context全局通知退出是无序的，无法保证数据不丢失。
cancel()-context通知退出
执行如下，A/B/C同时退出，数据出现丢失
2023/01/07 23:23:59 Task A start!
2023/01/07 23:23:59 Task B start!
2023/01/07 23:23:59 Task C start!
complex_test.go:174: Out->1
complex_test.go:174: Out->2
complex_test.go:174: Out->3
complex_test.go:174: Out->4
complex_test.go:174: Out->5
complex_test.go:174: Out->6
complex_test.go:161: Notify to stop!!!
2023/01/07 23:24:18 Task C revice exit signal!
complex_test.go:174: Out->7
close(dataChanInput)通知退出
执行如下，A/B/C逐级依次关闭，数据没有丢失
2023/01/07 23:20:18 Task A start!
2023/01/07 23:20:18 Task B start!
2023/01/07 23:20:18 Task C start!
complex_test.go:174: Out->1
complex_test.go:174: Out->2
complex_test.go:174: Out->3
complex_test.go:174: Out->4
complex_test.go:174: Out->5
complex_test.go:174: Out->6
complex_test.go:161: Notify to stop!!!
complex_test.go:174: Out->7
2023/01/07 23:20:37 Task A revice data channel close signal!
complex_test.go:174: Out->8
2023/01/07 23:20:39 Task B revice data channel close signal!
2023/01/07 23:20:41 Task C revice data channel close signal!
complex_test.go:174: Out->9
参考
演示代码 https://gitee.com/wenzhou1219/go-in-prod/tree/master/task_stop
来源：https://blog.csdn.net/wenzhou1219/article/details/128857554
0
投稿猜你喜欢
- 一个完整的数据挖掘模型，最后都要进行模型评估，对于二分类来说，AUC，ROC这两个指标用到最多，所以 利用sklearn里面相应的函数进行模
- 思考题假设有一张订单表 order，主要包含了主键订单编码 order_no、订单状态 status、提交时间 create_time 等列
- 目录一、Python 中的语句1.1什么是声明？1.2什么是表达式1.3简单的赋值语句1.4增强赋值语句二、Python 中的多行语句2.1
- 1.django自定义字段类型,实现非主键字段的自增# -*- encoding: utf-8 -*-from django.db.mode
mysql中写判断语句的方法：方法一.CASE函数case函数语法：CASE conditionWHEN value1 THEN retur- MySQL prepare语法： PREPARE statement_name FROM preparable_SQL_statement;
- 原文链接：https://blog.csdn.net/Fairy_Nan/article/details/105914203HDF也是一种自
- 问题描述：在画时间序列ACF时，调用from statsmodels.graphics.tsaplots import plot_acf,
- 目录prometheus通过exporter监控mysql，并用grafana图表展示1、测试机器 2、配置mysql host0
- 随机生成四位数验证码，包括汉字，数字，英文大小写。1.Servlet类package servlet;import java.awt.Bas
- 首先，如果你还没有对yield有个初步分认识，那么你先把yield看做“return”，这个是直观的
- Sometimes we have need to interact with an application,for examp
- 一、安装openpyxl模块Python操作excel主要用到了openpyxl模块，按win+R打开cmd，在里面输入pip3 insta
- 本文实例讲述了Python使用pickle模块报错EOFError Ran out of input的解决方法。分享给大家供大家参考，具体如
- 以下是它们的共同点： 1. 关于左右表的概念。左表指的是在SQL语句中排在left join左边的表，右表指的是排在left join右边的
- 1. 代码完整的源代码：import torchfrom torch import nn# 定义一个LSTM模型class LSTM(nn.
- 目录1.自动移动鼠标，以便Skype / Lynk显示你在工作中处于活动状态2.使用Selenium自动化网站登录过程3.自动文件备份4.自
- 之前一直使用QtCreator，在设计界面时非常方便，python早就集成了Qt模块，在python中以pyQt的包存在，目前常用的是pyQ
- 1.figure语法及操作(1)figure语法说明figure(num=None, figsize=None, dpi=None, fac
- 为什么会讲 MRO？在讲多继承的时候，有讲到， 当继承的多个父类拥有同名属性、方法，子类对象调用该属性、方法时会调用哪个父类的属性、方法呢？
