在通道关闭时向goroutines发出停止信号

Question

我从两个通道中获得了多个选择的goroutine:一个chan提供数据，一个chan提供信号（类似于done/quit通道）。

我使用signals通道来捕获信号（杀死）并优雅地关闭Goroutines。

package a

import "sync"

WorkChan := make(chan int)
QuitChan := make(chan struct{})

func Stop() {
        fmt.Println("Stop called, closing channel")
        close(QuitChan)
}

func Work(wg *sync.WaitGroup) {
    var item int
    for {
        select {
        case item = <- WorkChan:
            ... processing
        case <- QuitChan:
            wg.Done()
            return
        }
    }
}

package b

import (
    "os/signal"
    "os"
    "syscal"
    "a"
)

func Signal() {

    sChan := make(chan os.Signal, 1)
    signal.Notify(signalChan, syscall.SIGTERM, syscall.SIGINT)

    for {
        s := <-sChan
        switch s {
        case os.Interrupt, syscall.SIGTERM:
            a.Stop()
        }
    }
}

package main

import (
    "a"
    "b"
    "sync"
)

func main() {

    var wg sync.WaitGroup

    go b.Signal()

    wg.Add(1) // for simplicity; actual code start multiple goroutines of Work
    go a.Work(&wg)

    // wait until work is done
    wg.Wait()
    fmt.Println("Done.")
}

当我终止正在运行的进程时，我会看到来自quit的打印消息。我预计，一旦通道关闭，goroutines将selectthequitchancase并返回。

但他们不停地奔跑；它们继续处理workchan中的项。好像被忽视了。我在这里漏掉了什么？频道不是关闭了吗？怎么还开着？

Answer 1

首先，我认为你应该做一个简单的测试，并通过它。让别人了解你的问题会更有帮助。

我改变了你的代码，让它读起来像go代码，而不是其他语言。现在奏效了。

在你的代码中，有一些错误，我把它标记为错误注释。有些是语法错误，比如创建workchan。有些是类型错误。

package a

import (
    "fmt"
    "sync"
)

// ERROR: can not do make in global
var WorkChan chan int
var QuitChan chan struct{}

// Create chan when init
func init() {
    fmt.Println("Init a")
    WorkChan = make(chan int)
    QuitChan = make(chan struct{})
}

func Stop() {
    fmt.Println("Stop called, closing quit channel")
    close(QuitChan)
}

// Close the work channel where you send date
func Start(wg *sync.WaitGroup) {
    i := 0
    for {
        select {
        case <-QuitChan:
            fmt.Println("Closing work chan")
            close(WorkChan)
            wg.Done()
            return
        default:
            WorkChan <- i
            i++
        }
    }
}

// Work will exit when workchan closed
func Work(wg *sync.WaitGroup) {
    for item := range WorkChan {
        fmt.Printf("Receive %d\n", item)
    }
    wg.Done()
    fmt.Println("Work exit")
}

package b

import (
    "github.com/shitaibin/awesome/a"
    "os"
    "os/signal"
    "syscall"
)

func Signal() {

    sChan := make(chan os.Signal, 1)
    signal.Notify(sChan, syscall.SIGTERM, syscall.SIGINT) // ERROR

    for {
        s := <-sChan
        switch s {
        case os.Interrupt, syscall.SIGTERM:
            a.Stop()
            return // should return free resource
        }
    }
}

package main

import (
    "fmt"
    "github.com/shitaibin/awesome/a"
    "github.com/shitaibin/awesome/b"
    "sync"
)

func main() {

    var wg sync.WaitGroup

    go b.Signal()

    wg.Add(1)      // for simplicity; actual code start multiple goroutines of Work
    go a.Work(&wg) // ERROR: pointer

    wg.Add(1)
    go a.Start(&wg) // Send data and close channel when stop

    // wait until work is done
    wg.Wait()
    fmt.Println("Done.")
}

结果

// omit
Receive 87028
Receive 87029
Receive 87030
Receive 87031
Receive 87032
Receiv^C101    <---- send signal here
Receive 87102
Receive 87103
Receive 87104
Receive 87105
Receive 87106
Receive 87107
Receive 87108
Receive 87109
Receive 87110
Stop called, closing quit channel
Receive 87111
Receive 87112
Closing work chan
Work exit
Done.