Go：一个频道有多个听众

Question

我对 Go 还很陌生，如果主题错误，我很抱歉，但我希望你能理解我的问题。我想通过通道将事件处理到不同的 go 例程。这是一些示例代码

type Event struct {

    Host string

    Command string

    Output string

}

var (

    incoming        = make(chan Event)

)

func processEmail(ticker* time.Ticker) {

    for {

        select {

        case t := <-ticker.C:

            fmt.Println("Email Tick at", t)

        case e := <-incoming:

            fmt.Println("EMAIL GOT AN EVENT!")

            fmt.Println(e)

        }

    }

}

func processPagerDuty(ticker* time.Ticker) {

    for {

        select {

        case t := <-ticker.C:

            fmt.Println("Pagerduty Tick at", t)

        case e := <-incoming:

            fmt.Println("PAGERDUTY GOT AN EVENT!")

            fmt.Println(e)

        }

    }

}

func main() {

    err := gcfg.ReadFileInto(&cfg, "dispatch-api.cfg")

    if err != nil {

        fmt.Printf("Error loading the config")

    }

    ticker := time.NewTicker(time.Second * 10)

    go processEmail(ticker)

    ticker := time.NewTicker(time.Second * 1)

    go processPagerDuty(ticker)

}

func eventAdd(r render.Render, params martini.Params, req *http.Request) {

    // create an event now

    e := Event{Host: "web01-east.domain.com", Command: "foo", Output: "bar"}

    incoming <- e

}

所以股票代码事件工作只是创建。当我发出一个 API 调用来创建一个事件时，我只是从 processEmail 函数中获得输出。它首先调用的任何 go 例程都将通过通道获取事件。

有没有办法让两个函数都得到那个事件？

Answer 1

您可以使用扇入和扇出（来自 Rob Pike 的演讲）：package mainfunc main() {&nbsp; &nbsp; // feeders - feeder1, feeder2 and feeder3 are used to fan in&nbsp; &nbsp; // data into one channel&nbsp; &nbsp; go func() {&nbsp; &nbsp; &nbsp; &nbsp; for {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; select {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; case v1 := <-feeder1:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; mainChannel <- v1&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; case v2 := <-feeder2:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; mainChannel <- v2&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; case v3 := <-feeder3:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; mainChannel <- v3&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; }()&nbsp; &nbsp; // dispatchers - not actually fan out rather dispatching data&nbsp; &nbsp; go func() {&nbsp; &nbsp; &nbsp; &nbsp; for {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; v := <-mainChannel&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // use this to prevent leaking goroutines&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // (i.e. when one consumer got stuck)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; done := make(chan bool)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; go func() {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; consumer1 <- v&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; done <- true&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; go func() {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; consumer2 <- v&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; done <- true&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; go func() {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; consumer3 <- v&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; done <- true&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; <-done&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; <-done&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; <-done&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; }()&nbsp; &nbsp; // or fan out (when processing the data by just one consumer is enough)&nbsp; &nbsp; go func() {&nbsp; &nbsp; &nbsp; &nbsp; for {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; v := <-mainChannel&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; select {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; case consumer1 <- v:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; case consumer2 <- v:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; case consumer3 <- v:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; }()&nbsp; &nbsp; // consumers(your logic)&nbsp; &nbsp; go func() { <-consumer1 /* using the value */ }()&nbsp; &nbsp; go func() { <-consumer2 /* using the value */ }()&nbsp; &nbsp; go func() { <-consumer3 /* using the value */ }()}type payload intvar (&nbsp; &nbsp; feeder1 = make(chan payload)&nbsp; &nbsp; feeder2 = make(chan payload)&nbsp; &nbsp; feeder3 = make(chan payload)&nbsp; &nbsp; mainChannel = make(chan payload)&nbsp; &nbsp; consumer1 = make(chan payload)&nbsp; &nbsp; consumer2 = make(chan payload)&nbsp; &nbsp; consumer3 = make(chan payload))

Answer 2

通道是一种点对点通信方法，而不是广播通信方法，所以不，如果不做一些特殊的事情，您无法同时获得两个函数来获取事件。您可以为两个 goroutine 设置单独的通道并将消息发送到每个通道。这可能是最简单的解决方案。或者，您可以让一个 goroutine 向下一个发出信号。据我所知，Go 有两种机制来进行广播信令。一是关闭通道。但这只能工作一次。另一种是使用sync.Cond锁。这些使用起来比较棘手，但可以让您有多个 goroutines 被单个事件唤醒。如果我是你，我会选择第一个选项，将事件发送到两个不同的频道。这似乎很好地映射了问题。