通道和正常关机死锁

Question

运行以下程序并运行CTRL + C，例程在尝试发送到通道但例程已关闭时被阻止。什么是更好的并发设计来解决这个问题？handleprocess已编辑程序以应用此处建议的规则来描述问题，https://stackoverflow.com/a/66708290/4106031package mainimport (    "context"    "fmt"    "os"    "os/signal"    "sync"    "syscall"    "time")func process(ctx context.Context, c chan string) {    fmt.Println("process: processing (select)")    for {        select {        case <-ctx.Done():            fmt.Printf("process: ctx done bye\n")            return        case i := <-c:            fmt.Printf("process: received i: %v\n", i)        }    }}func handle(ctx context.Context, readChan <-chan string) {    c := make(chan string, 1)    wg := &sync.WaitGroup{}    wg.Add(1)    go func() {        process(ctx, c)        wg.Done()    }()    defer wg.Wait()    for i := 0; ; i++ {        select {        case <-ctx.Done():            fmt.Printf("handle: ctx done bye\n")            return        case i := <-readChan:            fmt.Printf("handle: received: %v\n", i)            fmt.Printf("handle: sending for processing: %v\n", i)            // suppose huge time passes here            // to cause the issue we want to happen            // we want the process() to exit due to ctx            // cancellation before send to it happens, this creates deadlock            time.Sleep(5 * time.Second)            // deadlock            c <- i        }    }}func main() {    wg := &sync.WaitGroup{}    ctx, cancel := context.WithCancel(context.Background())    defer cancel()    readChan := make(chan string, 10)    wg.Add(1)    go func() {        defer wg.Done()        for i := 0; ; i++ {            select {            case <-ctx.Done()                fmt.Printf("read: ctx done bye\n")                return            case readChan <- fmt.Sprintf("%d", i):                fmt.Printf("read: sent msg: %v\n", i)            }        }    }

Answer 1

分步回顾

无论你怎么想，总是取消上下文。

ctx, cancel := context.WithCancel(context.Background())

defer cancel()

不要说。启动例程后添加

    wg.Add(1)

    go handle(ctx, wg)

不要稀疏地消耗等待组

    wg.Add(1)

    go func() {

        handle(ctx)

        wg.Done()

    }()

不要在具有默认大小写的通道上循环。只需从中读取并让它解锁

    <-sigterm

    fmt.Printf("SIGTERM signal received\n")

主从不阻塞信号，主块在处理例程上。信令应该只做信令，即取消上下文。

    go func() {

        sigterm := make(chan os.Signal, 1)

        signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)

        <-sigterm

        fmt.Printf("SIGTERM signal received\n")

        cancel()

    }()

可以检查通道写入时的上下文取消。

        select {

        case <-ctx.Done():

            fmt.Printf("process: ctx done bye\n")

            return

        case c <- fmt.Sprintf("%d", i):

            fmt.Printf("handled: sent to channel: %v\n", i)

        }

不要花时间。睡眠，您无法用它来测试上下文取消。

        select {

        case <-ctx.Done():

            fmt.Printf("process: ctx done bye\n")

            return

        case <-time.After(time.Second * 5):

        }

因此，应用了各种规则的完整修订版代码为我们提供了

package main

import (

    "context"

    "fmt"

    "os"

    "os/signal"

    "sync"

    "syscall"

    "time"

)

func process(ctx context.Context, c chan string) {

    fmt.Println("process: processing (select)")

    for {

        select {

        case <-ctx.Done():

            fmt.Printf("process: ctx done bye\n")

            return

        case msg := <-c:

            fmt.Printf("process: got msg: %v\n", msg)

        }

    }

}

func handle(ctx context.Context) {

    c := make(chan string, 3)

    wg := &sync.WaitGroup{}

    wg.Add(1)

    go func() {

        process(ctx, c)

        wg.Done()

    }()

    defer wg.Wait()

    for i := 0; ; i++ {

        select {

        case <-ctx.Done():

            fmt.Printf("process: ctx done bye\n")

            return

        case <-time.After(time.Second * 5):

        }

        select {

        case <-ctx.Done():

            fmt.Printf("process: ctx done bye\n")

            return

        case c <- fmt.Sprintf("%d", i):

            fmt.Printf("handled: sent to channel: %v\n", i)

        }

    }

}

func main() {

    wg := &sync.WaitGroup{}

    ctx, cancel := context.WithCancel(context.Background())

    defer cancel()

    wg.Add(1)

    go func() {

        handle(ctx)

        wg.Done()

    }()

    go func() {

        sigterm := make(chan os.Signal, 1)

        signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)

        <-sigterm

        fmt.Printf("SIGTERM signal received\n")

        cancel()

    }()

    wg.Wait()

}

关于退出条件还有更多要讲的，但这取决于要求。

Answer 2

如前所述 https://stackoverflow.com/a/66708290/4106031 此更改为我解决了问题。也感谢mh-cbon的规则！