«

golang协程关闭实例分析

时间:2024-7-31 08:55     作者:韩俊     分类: Go语言


本篇内容主要讲解“golang协程关闭实例分析”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“golang协程关闭实例分析”吧!

场景

结合如下典型场景,主进程中起多个协程,这些协程会

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
", 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
", taskNo, val)

            // 关闭exit通道时,通知退出
        case <-exitChan:
            log.Printf("Task %d  revice exitChan signal!
", 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
", taskNo, val)

        // ctx取消或超时,通知退出
        case <-ctx.Done():
            log.Printf("Task %d  revice exit signal!
", 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!
", taskName)

    go func() {
        defer close(out)

        for {
            select {
            case data, ok := <-dataChan:
                if !ok {
                    log.Printf("Task %s  revice data channel close signal!
", taskName)
                    return
                }

                // do something
                out <- data
            case <-ctx.Done():
                log.Printf("Task %s  revice exit signal!
", 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
", taskNo)

            for {
                select {
                case _, ok := <-dataChan:
                    if !ok {
                        t.Logf("Task %d notify to stop
", 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
")
                    return
                }
            }
        }
    }()

    // 通知退出
    go func() {
        time.Sleep(3 * time.Second)
        close(dataChan)
    }()

    dataChan <- 1

    // 等待退出完成
    for data := range out {
        t.Logf("%d
", 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
")
                    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!
", taskName)
    f(ctx)

    for {
        select {

        case <-ctx.Done():
            log.Printf("Task %s  revice exit signal!
", 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!
", taskName)

    go func() {
        defer close(out)

        for {
            select {
            case data, ok := <-dataChan:
                if !ok {
                    log.Printf("Task %s  revice data channel close signal!
", taskName)
                    return
                }

                time.Sleep(2 * time.Second)
                out <- data
            case <-ctx.Done():
                log.Printf("Task %s  revice exit signal!
", 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

标签: golang

热门推荐