首先将等待 go-routines 和donechannel 分开。


使用 async.WaitGroup来协调 goroutine。


func main() {

    wait := &sync.WaitGroup{}

    N := 3


    wait.Add(N)

    for i := 1; i <= N; i++ {

        go goFunc(wait, i, true)

    }


    wait.Wait()

    fmt.Println(`Exiting main`)

}

每个 goroutine 将如下所示:


// code for the actual goroutine

func goFunc(wait *sync.WaitGroup, i int, closer bool) {

    defer wait.Done()

    defer fmt.Println(`Exiting `, i)


    T := time.Tick(time.Duration(100*i) * time.Millisecond)

    for {

        select {

        case <-T:

            fmt.Println(`Tick `, i)

            if closer {

                return

            }

        }

    }

}

(https://play.golang.org/p/mDO4P56lzBU)


我们的 main 函数在退出之前成功地等待 goroutines 退出。每个 goroutine 都在关闭自己,我们想要一种同时取消所有 goroutine 的方法。


我们将使用chan, 并利用从频道接收的这一特性:


引用:关闭通道上的接收操作总是可以立即进行,在接收到任何先前发送的值之后产生元素类型的零值。(https://golang.org/ref/spec#Receive_operator)


我们修改我们的 goroutine 来检查 CLOSE:


func goFunc(wait *sync.WaitGroup, i int, closer bool, CLOSE chan struct{}) {

    defer wait.Done()

    defer fmt.Println(`Exiting `, i)


    T := time.Tick(time.Duration(100*i) * time.Millisecond)

    for {

        select {

        case <-CLOSE:

            return

        case <-T:

            fmt.Println(`Tick `, i)

            if closer {

                close(CLOSE)

            }

        }

    }

}

然后我们改变我们的func main,让它通过 CLOSE 通道,我们将设置closer变量,以便只有我们的最后一个 goroutine 会触发关闭:


func main() {

    wait := &sync.WaitGroup{}

    N := 3

    CLOSE := make(chan struct{})


    // Launch the goroutines

    wait.Add(N)

    for i := 1; i <= N; i++ {

        go goFunc(wait, i, i == N, CLOSE)

    }


    // Wait for the goroutines to finish

    wait.Wait()

    fmt.Println(`Exiting main`)

}

(https://play.golang.org/p/E91CtRAHDp2)


现在看起来一切正常。


但事实并非如此。并发很难。这段代码中潜伏着一个错误,正等着在生产中咬你。让我们浮出水面。


更改我们的示例,以便每个goroutine 都将关闭:


func main() {

    wait := &sync.WaitGroup{}

    N := 3

    CLOSE := make(chan struct{})


    // Launch the goroutines

    wait.Add(N)

    for i := 1; i <= N; i++ {

        go goFunc(wait, i, true /*** EVERY GOROUTINE WILL CLOSE ***/, CLOSE)

    }


    // Wait for the goroutines to finish

    wait.Wait()

    fmt.Println(`Exiting main`)

}

更改 goroutine 以便在关闭之前需要一段时间。我们希望两个 goroutine 同时关闭:


// code for the actual goroutine

func goFunc(wait *sync.WaitGroup, i int, closer bool, CLOSE chan struct{}) {

    defer wait.Done()

    defer fmt.Println(`Exiting `, i)


    T := time.Tick(time.Duration(100*i) * time.Millisecond)

    for {

        select {

        case <-CLOSE:

            return

        case <-T:

            fmt.Println(`Tick `, i)

            if closer {

                /*** TAKE A WHILE BEFORE CLOSING ***/

                time.Sleep(time.Second)

                close(CLOSE)

            }

        }

    }

}



(https://play.golang.org/p/YHnbDpnJCks)


我们崩溃:


Tick  1

Tick  2

Tick  3

Exiting  1

Exiting  2

panic: close of closed channel


goroutine 7 [running]:

main.goFunc(0x40e020, 0x2, 0x68601, 0x430080)

    /tmp/sandbox558886627/prog.go:24 +0x2e0

created by main.main

    /tmp/sandbox558886627/prog.go:38 +0xc0


Program exited: status 2.

虽然关闭通道上的接收立即返回,但您无法关闭关闭的通道。


我们需要一点协调。我们可以用 async.Mutex和 abool来表示我们是否关闭了通道。让我们创建一个结构来执行此操作:


type Close struct {

    C chan struct{}

    l sync.Mutex

    closed bool

}


func NewClose() *Close {

    return &Close {

        C: make(chan struct{}),

    }

}


func (c *Close) Close() {

    c.l.Lock()

    if (!c.closed) {

        c.closed=true

        close(c.C)

    }

    c.l.Unlock()

}

重写我们的 gofunc 和我们的 main 以使用我们新的 Close 结构,我们很高兴: https: //play.golang.org/p/eH3djHu8EXW


并发的问题在于,您总是需要想知道如果另一个“线程”在代码中的其他任何地方会发生什么。