go - 如何有效地将 channel 链接在一起？

Question

我正在尝试在 golang 中创建消息中心。消息通过在 map[uint32]chan []float64 中持续存在的不同 channel 获取。我在 map 上无限循环并检查 channel 是否有消息。如果有，我将它与传入 channel 的 id 一起写入公共(public)客户端的写入 channel 。它工作正常，但使用所有 CPU，并且其他进程受到限制。

更新: map 中的项目由另一个函数动态添加和删除。

我想通过 Docker 限制此应用的 CPU，但也许有更优雅的路径？

我的代码:

    func (c *Client) accumHandler() {

    for !c.stop {
        c.channels.Range(func(key, value interface{}) bool {

            select {
            case message := <-value.(chan []float64):
                mess := map[uint32]interface{}{key.(uint32): message}

                select {
                case c.send <- mess:

                }

            default:

            }
            return !c.stop
        })
    }
}

Answer 1

如果我没看错的话，您似乎正试图将一组 float 连同 channel 标识符一起传递到公共(public) channel 。我假设您这样做是为了将多个 channel 传递给不同的发布者，但这样您只需为您的消费者跟踪一个 channel 。

事实证明，您无需遍历 channel 即可查看它何时输出值。您可以在 goroutine 中将 channel 链接在一起。因此，不需要忙等待。这样的事情将适合您的目的(同样，如果我没看错的话)。寻找绕过繁忙循环的全部大写注释。 Link to playground .

var num_publishes = 3

func main() {
  num_publishers := 10
  single_consumer := make(chan []float64)

  for i:=0;i<num_publishers;i+=1 {
    c := make(chan []float64)

    // connect channel to your single consumer channel
    go func() { for { single_consumer <- <-c } }() // THIS IS PROBABLY WHAT YOU DIDN'T KNOW ABOUT

    // send the channel to the publisher
    go publisher(c, i*100)
  }

  // dumb consumer example
  for i:=0;i<num_publishers*num_publishes;i+=1 {
    fmt.Println(<-single_consumer)
  }
}

func publisher(c chan []float64, publisher_id int) {
  dummy := []float64{
    float64(publisher_id+1),
    float64(publisher_id+2),
    float64(publisher_id+3),
  }
  for i:=0;i<num_publishes;i+=1 {
    time.Sleep(time.Duration(rand.Intn(10000)) * time.Millisecond)
    c <- dummy
  }
}