go并发日记·避免goroutine泄漏/实现协程可控

亦凉 2023-05-22 06:45 7阅读 0赞

## 骐骥一跃，不能十步；驽马十驾，功在不舍。 ##

**go并发之美·干货好文合集：**

[go并发之美·redis篇·接口幂等性设计实现][go_redis]

[go并发之美·redis篇·实现实时排行榜][go_redis 1]

[go并发日记·多goroutine并发安全的操作共享变量][go_goroutine]

[go并发日记·常用并发场景中的并发控制思路与方案][go]

[go并发日记·避免goroutine泄漏/实现协程可控][go_goroutine 1]

[go并发日记·深入理解go语言channel与goroutine阻塞][go_go_channel_goroutine]

[go并发日记·基于channel优雅的进行并发控制][go_channel]

[go并发日记·WaitGroup并发控制探索及实际场景运用][go_WaitGroup]

[channel死锁的情况分析][channel]

[go并发之美·B操作伴随A操作(B和A同步)][go_B_A_B_A]

[go多函数并发通用工具的设计与实现][go 1]

干货的不能再干货了。

**目录**

不寻常的close

goroutine内部逻辑来触发控制结束goroutine

姿势1

姿势2

姿势3

外部条件触发控制结束goroutine

多个协程，某个goroutine逻辑内部报错需要停止时，需要所有goroutine都结束

多个协程A、B、C，其中A、B协程需要停止时不影响其他(C)协程：并行式

多个协程A、B、C，其中A、B协程需要停止时不影响其他(C)协程：继承式

总结

--------------------

## 不寻常的close ##

func main() {
    	fmt.Println("1当前运行的goroutine: ", runtime.NumGoroutine())
    	ch := make(chan bool)
    	go func() {
    		fmt.Println("2当前运行的goroutine: ", runtime.NumGoroutine())
    		ch <- true
    		close(ch) // 关闭即信号，如不关闭 fatal error: all goroutines are asleep - deadlock!
    	}()
    
    	for value := range ch { 
    		fmt.Println(value)
    	}
         // 通道关闭后&读完后 range才会结束，后面的代码才会执行
    	fmt.Println("结束")
    	fmt.Println("3当前运行的goroutine: ", runtime.NumGoroutine())
    }
    
    控制台(close时)：
    
    1当前运行的goroutine:  1
    2当前运行的goroutine:  2
    true
    结束
    3当前运行的goroutine:  1
    
    Process finished with exit code 0

## goroutine内部逻辑来触发控制结束goroutine ##

实际背景：逻辑报错了需要停止任务并结束goroutine

### 姿势1 ###

一般goroutine+channel方式

func main() {
    	fmt.Println("1当前运行的goroutine: ", runtime.NumGoroutine())//一开始只有main线
    	signCh := make(chan struct{}) // 信号通知&阻塞main线
    	// 任务A：doSomeThing
    	go func() {
    		defer close(signCh)  //重要性：一般重要(因为下面是单次读取即可解除阻塞)
    		fmt.Println("2当前运行的goroutine: ", runtime.NumGoroutine())
    		for i := 0; ; i++ {
    			fmt.Println("任务A进行中：", i)
    			if i == 2 { // 模拟出现报错或需要停止的情况
    				signCh <- struct{}{}
    				return //重要性：很重要
    			}
    			time.Sleep(time.Second * 1)
    		}
    	}()
    
    	<-signCh
    	defer func() {
    		fmt.Println("3当前运行的goroutine: ", runtime.NumGoroutine())
    		fmt.Println("main线退出。")
    	}()
    
    }
    
    控制台：
    1当前运行的goroutine:  1
    2当前运行的goroutine:  2
    任务A进行中： 0
    任务A进行中： 1
    任务A进行中： 2
    3当前运行的goroutine:  1
    main线退出。
    
    Process finished with exit code 0

### 姿势2 ###

基于context+channel，报错时执行cancel()同时停止任务，以信号channel解除阻塞

func main() {
    	fmt.Println("1当前运行的goroutine: ", runtime.NumGoroutine())
    	signCh := make(chan struct{})
    	ctx, cancel := context.WithCancel(context.Background())
    	// 任务A：doSomeThing
    	go func(c context.Context) {
    		for {
    			select {
    			case <-c.Done():
    				fmt.Println("任务A停止")
    				signCh <- struct{}{}
    				return
    			default:
    				for i := 0; ; i++ {
    					fmt.Println("任务A：", i)
    					if i == 2 {
    						cancel()
    						break
    					}
    					time.Sleep(time.Second * 1)
    					fmt.Println("2当前运行的goroutine: ", runtime.NumGoroutine())
    				}
    			}
    		}
    	}(ctx)
    
    	<-signCh
    
    	defer func() {
    		fmt.Println("3当前运行的goroutine: ", runtime.NumGoroutine()) // 1
    	}()
    }
    
    控制台：
    1当前运行的goroutine:  1
    任务A： 0
    2当前运行的goroutine:  2
    任务A： 1
    2当前运行的goroutine:  2
    任务A： 2
    任务A停止
    3当前运行的goroutine:  1
    
    Process finished with exit code 0

如果你想写的代码更多一点，看起来更复杂一点，那么也可以瞧瞧方式3~哈哈(小生不才，认为代码的变换和综合运用很有趣)

### 姿势3 ###

基于context包，把for select用烂，多写点代码玩玩

func main() {
    	fmt.Println("1当前运行的goroutine: ", runtime.NumGoroutine())
    	signCh := make(chan struct{})
    	ctx, cancel := context.WithCancel(context.Background())
    	// 任务A：doSomeThing
    	go func(c context.Context) {
    		for i := 0; ; i++ {
    			if i == 2 { // 模拟出现报错或需要停止的情况
    				fmt.Println("任务A准备停止...")
    				cancel() // 发出取消指令
    				break    // 停止当前任务
    			}
    			fmt.Println("任务A进行中：", i)
    			time.Sleep(time.Second * 1)
    		}
    	}(ctx)
    
    	go func(c context.Context) { // 专门用来读取停止信号
    		for {
    			select {
    			case <-c.Done():
    				fmt.Println("任务A停止。所有goroutine即将结束...")
    				signCh <- struct{}{} // 发送信号，解除阻塞main线
    				return
    			default:
    				fmt.Println("2当前运行的goroutine: ", runtime.NumGoroutine())
    				time.Sleep(time.Second * 2)
    			}
    		}
    	}(ctx)
    
        //停止for-select循环方式也可移步https://lan6193.blog.csdn.net/article/details/101208252
    	stop := false
    	for !stop {    
    		select {
    		case <-signCh:
    			stop = true
    		default:
    		}
    	}
    
    	defer func() {
    		time.Sleep(time.Second * 2)   
    		fmt.Println("3当前运行的goroutine: ", runtime.NumGoroutine()) // 1
    	}()
    }
    
    控制台：
    1当前运行的goroutine:  1
    任务A进行中： 0
    2当前运行的goroutine:  3
    任务A进行中： 1
    2当前运行的goroutine:  3
    任务A准备停止...
    任务A停止。所有goroutine即将结束...
    3当前运行的goroutine:  1
    
    Process finished with exit code 0

## 外部条件触发控制结束goroutine ##

方式：使用context的cancel函数

func main() {
    	fmt.Println("1当前运行的goroutine: ", runtime.NumGoroutine())
    	ctx, cancel := context.WithCancel(context.Background())
    	// 创建goroutine，传入ctx
    	go func(c context.Context) {
    		fmt.Println("2当前运行的goroutine: ", runtime.NumGoroutine())
    		for {
    			select {
    			case <-c.Done():
    				fmt.Println("任务A结束。")
    				return
    			default:
    				fmt.Println(" 任务A进行中...")
    				time.Sleep(time.Second * 1)
    			}
    		}
    	}(ctx)
    
    	time.Sleep(time.Second * 3)
    	fmt.Println("此处暂以延迟3秒来模拟需要停止goroutine的外部条件，准备停止：")
    	cancel() 
    	fmt.Println("------------")
    	defer func() {
            // 经本地测试，goroutine退出需要一定时间，为两秒时上面的goroutine退出成功。
    		time.Sleep(time.Second * 2)  
    		fmt.Println("3当前运行的goroutine: ", runtime.NumGoroutine()) // 1
    	}()
    }
    
    
    控制台：
    1当前运行的goroutine:  1
    2当前运行的goroutine:  2
     任务A进行中...
     任务A进行中...
     任务A进行中...
    此处暂以延迟3秒来模拟需要停止goroutine的外部条件，准备停止：
    ------------
    任务A结束。
    3当前运行的goroutine:  1
    
    Process finished with exit code 0

## 多个协程，某个goroutine逻辑内部报错需要停止时，需要所有goroutine都结束 ##

多个协程，有一个goroutine中的逻辑报错或需要停止，需要所有goroutine都结束

// 多个协程，有一个goroutine中的逻辑报错或需要停止，需要所有goroutine都结束
    func workM(ctx context.Context, ch chan struct{}, f context.CancelFunc) {
    	str := []string{"A", "B", "C"} //纯粹为了区分三个协程：A、B、C三个任务分别对应自己的协程
    	for i := range str {
    		go func(name string, c context.Context) {
    			for {
    				select {
    				case <-c.Done():
    					fmt.Println("协程" + name + "结束。")
    					return
    				default:
    					fmt.Println("任务\t" + name + "\t正在执行...")
    					time.Sleep(time.Second * 1)
                                          //模拟协程B开始执行B任务2秒后因内部逻辑报错，需要停止
    					if name == "B" { 
    						time.Sleep(time.Second * 2)
    						fmt.Println("B出错了，都准备给我停下！")
    						f()
    						ch <- struct{}{}
    					}
    				}
    			}
    		}(str[i], ctx)
    	}
    }
    
    func main() {
    	fmt.Println("1当前运行的goroutine: ", runtime.NumGoroutine())
    	signCh := make(chan struct{}) // 用来阻塞main线
    	ctx, cancel := context.WithCancel(context.Background())
    	workM(ctx, signCh, cancel)
    	stop := false
    	for !stop {
    		select {
    		case <-signCh:
    			stop = true
    		default:
    			fmt.Println("2当前运行的goroutine: ", runtime.NumGoroutine())
    			time.Sleep(time.Second * 3)
    		}
    	}
    
    	defer func() {
    		time.Sleep(time.Second * 1)
    		fmt.Println("3当前运行的goroutine: ", runtime.NumGoroutine()) // 1
    	}()
    }
    
    控制台：
    
    1当前运行的goroutine:  1
    2当前运行的goroutine:  4
    任务	C	正在执行...
    任务	B	正在执行...
    任务	A	正在执行...
    任务	A	正在执行...
    任务	C	正在执行...
    任务	C	正在执行...
    任务	A	正在执行...
    2当前运行的goroutine:  4
    B出错了，都准备给我停下！
    协程A结束。
    协程C结束。
    协程B结束。
    3当前运行的goroutine:  1
    
    Process finished with exit code 0

该需求的另外一种转移解决方式：其中某一个协程内部逻辑报错时往ch中发消息，将cancel()放在外部，select 中监控到ch中的值后触发执行cancel()，因此其他协程读到ctx.Done的值即可退出，相当于内-->外-->内，也能达到所有协程都退出的目的。

## 多个协程A、B、C，其中A、B协程需要停止时不影响其他(C)协程：并行式 ##

背景需求：多个协程，有一个goroutine中的逻辑报错或需要停止，其他goroutine仍然正常运行，只结束报错的这个goroutine

假设有A、B、C三个协程，需求是A报错时结束A和B协程，C依旧正常

// 此处以公共标志位决定每个协程处理的业务逻辑for循环是否停止，也可采用其他办法，比如同
    //一channel等，当然这些不重要，重要的是让A和B一起结束的手段
    var flag bool = false
    
    func workAA(ctxAB context.Context, name string, cancelAB context.CancelFunc) {
    	for {
    		select {
    		case <-ctxAB.Done():
    			fmt.Println("协程" + name + "结束。")
    			return
    		default:
    			for i := 0; ; i++ {
    				fmt.Println("任务"+name+"进行中：", i)
    				fmt.Println("2-1当前运行的goroutine: ", runtime.NumGoroutine())
    				time.Sleep(time.Second * 1)
    				//模拟协程A内部逻辑报错了，需要A、B一起停止；这里让A(其中之一都可以)先报错
    				if i == 3 {
    					fmt.Println("任务" + name + "结束。")
    					break
    				}
    			}
    			flag = true
    			cancelAB() // 通知协程A、B都要结束掉
    		}
    	}
    }
    
    func workBB(ctxAB context.Context, name string, cancelAB context.CancelFunc) {
    	for {
    		select {
    		case <-ctxAB.Done():
    			fmt.Println("协程" + name + "结束。")
    			return
    		default:
    			for !flag {
    				fmt.Println("任务" + name + "进行中.")
    				fmt.Println("2-2当前运行的goroutine: ", runtime.NumGoroutine())
    				time.Sleep(time.Second * 1)
    			}
    		}
    	}
    
    }
    
    func workCC(ctxC context.Context, name string, cancelC context.CancelFunc) {
    	for {
    		select {
    		case <-ctxC.Done():
    			fmt.Println("协程" + name + "结束。")
    			return
    		default:
    			for {  // 让C一直做它的事，C的业务场景不能被A和B关联停止
    				fmt.Println("任务" + name + "进行中.")
    				fmt.Println("2-3当前运行的goroutine: ", runtime.NumGoroutine())
    				time.Sleep(time.Second * 1)
    			}
    		}
    	}
    }
    
    func main() {
    	fmt.Println("1当前运行的goroutine: ", runtime.NumGoroutine())
    	ctxParent, _ := context.WithCancel(context.Background()) // 根上下文
    	ctxAB, cancelAB := context.WithCancel(ctxParent)
    	ctxC, cancelC := context.WithCancel(ctxParent)
    	go workAA(ctxAB, "A", cancelAB) //该goroutine对应为A协程
    	go workBB(ctxAB, "B", cancelAB) //该goroutine对应为B协程
    	go workCC(ctxC, "C", cancelC)   //该goroutine对应为C协程
    
    	time.Sleep(time.Second * 8) // 用延迟来更好观察A、B是否都结束了
    	defer func() {
    		fmt.Println("3当前运行的goroutine: ", runtime.NumGoroutine()) // 为2是正常的
    	}()
    }
    
    
    控制台：
    1当前运行的goroutine:  1
    任务A进行中： 0
    2-1当前运行的goroutine:  4
    任务B进行中.
    任务B进行中.
    2-2当前运行的goroutine:  4
    任务A进行中： 1
    2-1当前运行的goroutine:  4
    任务A进行中： 2
    2-1当前运行的goroutine:  4
    任务B进行中.
    2-2当前运行的goroutine:  4
    任务A进行中： 3
    2-1当前运行的goroutine:  4
    任务B进行中.
    2-2当前运行的goroutine:  4
    任务B进行中.
    2-2当前运行的goroutine:  4
    任务A结束。
    任务C进行中.
    2-3当前运行的goroutine:  4
    协程A结束。
    任务C进行中.
    2-3当前运行的goroutine:  3
    协程B结束。
    任务C进行中.
    2-3当前运行的goroutine:  2
    任务C进行中.
    2-3当前运行的goroutine:  2
    3当前运行的goroutine:  2
    
    Process finished with exit code 0

显然，达到了A和B一起结束的效果，并且C并不受影响！

## 多个协程A、B、C，其中A、B协程需要停止时不影响其他(C)协程：继承式 ##

package main
    
    import (
    	"context"
    	"fmt"
    	"runtime"
    	"time"
    )
    
    var flag bool = false
    
    func workAA(ctxAB context.Context, name string, cancelA context.CancelFunc) {
    	for {
    		select {
    		case <-ctxAB.Done():
    			fmt.Println("协程" + name + "结束。")
    			return
    		default:
    			for i := 0; ; i++ {
    				fmt.Println("任务"+name+"进行中：", i)
    				fmt.Println("2-1当前运行的goroutine: ", runtime.NumGoroutine())
    				time.Sleep(time.Second * 1)
    				//模拟协程A内部逻辑报错了，需要A、B一起停止；这里让A(其中之一都可以)先报错
    				if i == 3 {
    					fmt.Println("任务" + name + "结束。")
    					break
    				}
    				flag = true
    				cancelA() // 通知协程A、B都要结束掉
    			}
    
    		}
    	}
    }
    
    // 让B随着A结束，此处入参cancel我们不用它
    func workBB(ctxAB context.Context, name string, cancel context.CancelFunc) {
    	for {
    		select {
    		case <-ctxAB.Done():
    			fmt.Println("协程" + name + "结束。")
    			return
    		default:
    			for !flag {
    				fmt.Println("任务" + name + "进行中.")
    				fmt.Println("2-2当前运行的goroutine: ", runtime.NumGoroutine())
    				time.Sleep(time.Second * 1)
    			}
    			fmt.Println("任务" + name + "结束。")
    		}
    	}
    
    }
    
    func workCC(ctxC context.Context, name string, cancelC context.CancelFunc) {
    	for {
    		select {
    		case <-ctxC.Done():
    			fmt.Println("协程" + name + "结束。")
    			return
    		default:
    			for {
    				fmt.Println("任务" + name + "进行中.")
    				fmt.Println("2-3当前运行的goroutine: ", runtime.NumGoroutine())
    				time.Sleep(time.Second * 1)
    			}
    		}
    	}
    }
    
    func main() {
    	fmt.Println("1当前运行的goroutine: ", runtime.NumGoroutine())
    	ctxParent, _ := context.WithCancel(context.Background()) // 根上下文
    	ctxA, cancelA := context.WithCancel(ctxParent)
    	ctxB, _ := context.WithCancel(ctxA)
    	ctxC, cancelC := context.WithCancel(ctxParent) // B 继承A，让C 和 A平行
    	go workAA(ctxA, "A", cancelA)                  //该goroutine对应为A协程
    	go workBB(ctxB, "B", cancelA)                  //该goroutine对应为B协程
    	go workCC(ctxC, "C", cancelC)                  //该goroutine对应为C协程
    
    	time.Sleep(time.Second * 8) // 用延迟来更好观察A、B是否都结束了
    	defer func() {
    		fmt.Println("3当前运行的goroutine: ", runtime.NumGoroutine()) // 为2是正常的
    	}()
    }
    
    
    控制台
    1当前运行的goroutine:  1
    任务A进行中： 0
    2-1当前运行的goroutine:  4
    任务B进行中.
    任务C进行中.
    2-3当前运行的goroutine:  4
    任务A进行中： 1
    2-1当前运行的goroutine:  4
    任务B结束。
    协程B结束。
    任务C进行中.
    2-3当前运行的goroutine:  3
    任务A进行中： 2
    2-1当前运行的goroutine:  3
    任务C进行中.
    2-3当前运行的goroutine:  3
    任务A进行中： 3
    2-1当前运行的goroutine:  3
    任务C进行中.
    2-3当前运行的goroutine:  3
    任务A结束。
    协程A结束。
    任务C进行中.
    2-3当前运行的goroutine:  2
    任务C进行中.
    2-3当前运行的goroutine:  2
    任务C进行中.
    2-3当前运行的goroutine:  2
    3当前运行的goroutine:  2
    
    Process finished with exit code 0

结果同上一节的效果，达到了我们的目的。美滋滋！

## 总结 ##

业务场景A和B关联，在业务逻辑A终止时要结束场景B，则做一些操作让B跟随A，A停止则B也停止就可以了；而不让C一起停止，那么别关联C让C用另一个与其平行的ctx就行了(当然，C最后出错也要停止时，这时候执行自己的cancel就行了)，手段就是对ctx和cancel的控制。

未完待续！

如有任何问题欢迎讨论和指针！如果解决了你的疑惑，不妨点个赞哦，再会~

[go_redis]: https://lan6193.blog.csdn.net/article/details/108938119
[go_redis 1]: https://lan6193.blog.csdn.net/article/details/108931188
[go_goroutine]: https://lan6193.blog.csdn.net/article/details/105715373
[go]: https://lan6193.blog.csdn.net/article/details/108670428
[go_goroutine 1]: https://lan6193.blog.csdn.net/article/details/105559568
[go_go_channel_goroutine]: https://lan6193.blog.csdn.net/article/details/97528721
[go_channel]: https://lan6193.blog.csdn.net/article/details/101201408
[go_WaitGroup]: https://lan6193.blog.csdn.net/article/details/100887891
[channel]: https://lan6193.blog.csdn.net/article/details/97656420
[go_B_A_B_A]: https://lan6193.blog.csdn.net/article/details/108901339
[go 1]: https://lan6193.blog.csdn.net/article/details/103298105