99

这是@Jimt 编写的Go 中worker 和controller 模式的一个很好的例子,以回答“Golang 中是否有一些优雅的方式来暂停和恢复任何其他goroutine?

package main

import (
    "fmt"
    "runtime"
    "sync"
    "time"
)

// Possible worker states.
const (
    Stopped = 0
    Paused  = 1
    Running = 2
)

// Maximum number of workers.
const WorkerCount = 1000

func main() {
    // Launch workers.
    var wg sync.WaitGroup
    wg.Add(WorkerCount + 1)

    workers := make([]chan int, WorkerCount)
    for i := range workers {
        workers[i] = make(chan int)

        go func(i int) {
            worker(i, workers[i])
            wg.Done()
        }(i)
    }

    // Launch controller routine.
    go func() {
        controller(workers)
        wg.Done()
    }()

    // Wait for all goroutines to finish.
    wg.Wait()
}

func worker(id int, ws <-chan int) {
    state := Paused // Begin in the paused state.

    for {
        select {
        case state = <-ws:
            switch state {
            case Stopped:
                fmt.Printf("Worker %d: Stopped\n", id)
                return
            case Running:
                fmt.Printf("Worker %d: Running\n", id)
            case Paused:
                fmt.Printf("Worker %d: Paused\n", id)
            }

        default:
            // We use runtime.Gosched() to prevent a deadlock in this case.
            // It will not be needed of work is performed here which yields
            // to the scheduler.
            runtime.Gosched()

            if state == Paused {
                break
            }

            // Do actual work here.
        }
    }
}

// controller handles the current state of all workers. They can be
// instructed to be either running, paused or stopped entirely.
func controller(workers []chan int) {
    // Start workers
    for i := range workers {
        workers[i] <- Running
    }

    // Pause workers.
    <-time.After(1e9)
    for i := range workers {
        workers[i] <- Paused
    }

    // Unpause workers.
    <-time.After(1e9)
    for i := range workers {
        workers[i] <- Running
    }

    // Shutdown workers.
    <-time.After(1e9)
    for i := range workers {
        close(workers[i])
    }
}

但是这段代码也有一个问题:如果你想在退出workers时删除一个工作通道worker(),就会发生死锁。

如果你close(workers[i]),下次控制器写入它会导致恐慌,因为 go 无法写入关闭的通道。如果您使用一些互斥锁来保护它,那么它将被卡住,workers[i] <- Running因为worker它没有从通道读取任何内容并且写入将被阻塞,并且互斥锁将导致死锁。您也可以为通道提供更大的缓冲区作为解决方法,但这还不够好。

所以我认为解决这个问题的最好方法是worker()退出时关闭通道,如果控制器发现通道关闭,它将跳过它并且什么都不做。但是在这种情况下,我找不到如何检查通道是否已关闭。如果我尝试读取控制器中的通道,控制器可能被阻塞。所以我现在很困惑。

PS:恢复引发的恐慌是我尝试过的,但它会关闭引发恐慌的goroutine。在这种情况下,它将是控制器,所以没有用。

不过,我认为 Go 团队在下一个版本的 Go 中实现这个功能是有用的。

4

10 回答 10

86

没有办法编写一个安全的应用程序,您需要知道一个通​​道是否打开而不与它交互。

做你想做的事情的最好方法是使用两个渠道——一个用于工作,一个用于表明改变状态的愿望(以及如果这很重要,那么状态改变的完成)。

频道很便宜。复杂的设计重载语义不是。

[还]

<-time.After(1e9)

是一种非常令人困惑和不明显的写作方式

time.Sleep(time.Second)

保持简单,每个人(包括你)都能理解它们。

于 2013-04-21T02:19:40.800 回答
79

可以通过一种 hacky 的方式,通过恢复引发的恐慌来为尝试写入的通道完成。但是如果不读取读取通道,则无法检查它是否已关闭。

要么你会

  • 最终从中读取“真实”值 ( v <- c)
  • 读取“真”值和“未关闭”指示符 ( v, ok <- c)
  • 读取零值和“关闭”指示器 ( v, ok <- c)(示例
  • 将阻塞在永久读取的通道中 ( v <- c)

只有最后一个在技术上不会从频道中读取,但这没什么用。

于 2013-04-19T13:25:49.987 回答
8

我知道这个答案来得太晚了,我已经写了这个解决方案,Hacking Go run-time,它不安全,它可能会崩溃:

import (
    "unsafe"
    "reflect"
)


func isChanClosed(ch interface{}) bool {
    if reflect.TypeOf(ch).Kind() != reflect.Chan {
        panic("only channels!")
    }
    
    // get interface value pointer, from cgo_export 
    // typedef struct { void *t; void *v; } GoInterface;
    // then get channel real pointer
    cptr := *(*uintptr)(unsafe.Pointer(
        unsafe.Pointer(uintptr(unsafe.Pointer(&ch)) + unsafe.Sizeof(uint(0))),
    ))
    
    // this function will return true if chan.closed > 0
    // see hchan on https://github.com/golang/go/blob/master/src/runtime/chan.go 
    // type hchan struct {
    // qcount   uint           // total data in the queue
    // dataqsiz uint           // size of the circular queue
    // buf      unsafe.Pointer // points to an array of dataqsiz elements
    // elemsize uint16
    // closed   uint32
    // **
    
    cptr += unsafe.Sizeof(uint(0))*2
    cptr += unsafe.Sizeof(unsafe.Pointer(uintptr(0)))
    cptr += unsafe.Sizeof(uint16(0))
    return *(*uint32)(unsafe.Pointer(cptr)) > 0
}
于 2016-08-08T18:54:47.647 回答
2

我经常遇到多个并发 goroutine 的问题。

它可能是也可能不是一个好的模式,但我为我的工人定义了一个结构体,为工人状态定义了一个退出通道和字段:

type Worker struct {
    data chan struct
    quit chan bool
    stopped bool
}

然后你可以让一个控制器为工人调用一个停止函数:

func (w *Worker) Stop() {
    w.quit <- true
    w.stopped = true
}

func (w *Worker) eventloop() {
    for {
        if w.Stopped {
            return
        }
        select {
            case d := <-w.data:
                //DO something
                if w.Stopped {
                    return
                }
            case <-w.quit:
                return
        }
    }
}

这为您提供了一种很好的方法来完全停止您的工作人员,而不会出现任何挂起或产生错误,这在容器中运行时尤其有用。

于 2021-02-26T14:17:11.493 回答
2

好吧,你可以使用default分支来检测它,因为会选择一个关闭的通道,例如:下面的代码会选择default, channel, channel,第一个选择不会被阻塞。

func main() {
    ch := make(chan int)

    go func() {
        select {
        case <-ch:
            log.Printf("1.channel")
        default:
            log.Printf("1.default")
        }
        select {
        case <-ch:
            log.Printf("2.channel")
        }
        close(ch)
        select {
        case <-ch:
            log.Printf("3.channel")
        default:
            log.Printf("3.default")
        }
    }()
    time.Sleep(time.Second)
    ch <- 1
    time.Sleep(time.Second)
}

印刷

2018/05/24 08:00:00 1.default
2018/05/24 08:00:01 2.channel
2018/05/24 08:00:01 3.channel

请注意,请参阅@Angad 在此答案下的评论:

如果您使用的是缓冲通道并且它包含未读数据,则它不起作用

于 2018-05-24T08:01:45.363 回答
1

除了关闭它之外,您还可以将您的频道设置为 nil。这样你就可以检查它是否为零。

操场上的例子:https: //play.golang.org/p/v0f3d4DisCz

编辑:这实际上是一个糟糕的解决方案,如下一个示例所示,因为在函数中将通道设置为 nil 会破坏它: https: //play.golang.org/p/YVE2-LV9TOp

于 2020-06-28T08:14:24.927 回答
0
ch1 := make(chan int)
ch2 := make(chan int)
go func(){
    for i:=0; i<10; i++{
        ch1 <- i
    }
    close(ch1)
}()
go func(){
    for i:=10; i<15; i++{
        ch2 <- i
    }
    close(ch2)
}()
ok1, ok2 := false, false
v := 0
for{
    ok1, ok2 = true, true
    select{
        case v,ok1 = <-ch1:
        if ok1 {fmt.Println(v)}
        default:
    }
    select{
        case v,ok2 = <-ch2:
        if ok2 {fmt.Println(v)}
        default:
    }
    if !ok1 && !ok2{return}
    
}

}

于 2021-07-26T21:25:39.667 回答
-1

从文档中:

可以使用内置函数关闭通道。接收运算符的多值赋值形式报告是否在通道关闭之前发送了接收值。

https://golang.org/ref/spec#Receive_operator

Golang in Action 的示例显示了这种情况:

// This sample program demonstrates how to use an unbuffered
// channel to simulate a game of tennis between two goroutines.
package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

// wg is used to wait for the program to finish.
var wg sync.WaitGroup

func init() {
    rand.Seed(time.Now().UnixNano())
}

// main is the entry point for all Go programs.
func main() {
    // Create an unbuffered channel.
    court := make(chan int)
    // Add a count of two, one for each goroutine.
    wg.Add(2)
    // Launch two players.
    go player("Nadal", court)
    go player("Djokovic", court)
    // Start the set.
    court <- 1
    // Wait for the game to finish.
    wg.Wait()
}

// player simulates a person playing the game of tennis.
func player(name string, court chan int) {
    // Schedule the call to Done to tell main we are done.
    defer wg.Done()
    for {
        // Wait for the ball to be hit back to us.
        ball, ok := <-court
        fmt.Printf("ok %t\n", ok)
        if !ok {
            // If the channel was closed we won.
            fmt.Printf("Player %s Won\n", name)
            return
        }
        // Pick a random number and see if we miss the ball.
        n := rand.Intn(100)
        if n%13 == 0 {
            fmt.Printf("Player %s Missed\n", name)
            // Close the channel to signal we lost.
            close(court)
            return
        }

        // Display and then increment the hit count by one.
        fmt.Printf("Player %s Hit %d\n", name, ball)
        ball++
        // Hit the ball back to the opposing player.
        court <- ball
    }
}
于 2017-07-30T18:43:43.287 回答
-5

首先检查通道是否有元素更容易,这将确保通道处于活动状态。

func isChanClosed(ch chan interface{}) bool {
    if len(ch) == 0 {
        select {
        case _, ok := <-ch:
            return !ok
        }
    }
    return false 
}
于 2015-07-20T11:58:28.340 回答
-8

如果您收听此频道,您总是可以发现该频道已关闭。

case state, opened := <-ws:
    if !opened {
         // channel was closed 
         // return or made some final work
    }
    switch state {
        case Stopped:

但请记住,您不能两次关闭一个频道。这会引起恐慌。

于 2013-04-19T13:19:07.203 回答