我如何有效地将频道链接在一起？

Question

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

UPD：地图中的项目由另一个函数动态添加和删除。

我想通过 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

它正在吃掉你所有的 CPU，因为你不断地在字典中循环检查消息，所以即使没有消息要处理 CPU，或者至少没有线程或核心，也会耗尽。您需要阻止通道上的发送和接收！

我假设您这样做是因为您不知道会有多少个通道，因此不能在所有输入通道上只使用select。更好的模式是为您当前存储在字典中的每个输入通道启动一个单独的 goroutine。每个 goroutine 都应该有一个循环，在循环中它阻塞等待输入通道，并在接收到消息时阻塞发送到客户端的通道，该通道由所有人共享。

这个问题不完整，所以不能给出确切的代码，但是你会得到看起来像这样的 goroutine：

type Message struct {
    id uint32,
    message []float64
}

func receiverGoroutine(id uint32, input chan []float64, output chan Message) {
    for {
        message := <- input
        output <- Message{id: id, message: message}
    }
}

func clientGoroutine(c *Client, input chan Message) {
    for {
        message := <- input
        // do stuff
    }
}

（不过，您还需要添加一些“完成”频道）

在其他地方，您将使用如下代码开始它们：

clientChan := make(chan Message)
go clientGoroutine(client, clientChan)

for i:=0; i<max; i++ {
    go receiverGoroutine( (uint32)i, make(chan []float64, clientChan)
}

或者您可以只启动客户端例程，然后根据需要添加其他例程，而不是预先循环 - 取决于您的用例。

Answer 2

如果我正确读取卡片，您似乎正在尝试将浮点数组与通道标识符一起传递到公共通道。我假设您这样做是为了将多个渠道传递给不同的发布者，但是您只需为您的消费者跟踪一个渠道。

事实证明，您不需要遍历通道来查看它何时输出值。您可以在 goroutine 内将通道链接在一起。因此，不需要忙等待。这样的事情将适合您的目的（再次，如果我正确阅读卡片）。寻找所有大写的评论，以了解您的繁忙循环。 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
  }
}