从ch频道读取len(ch)消息可以吗？

Question

我正在运行单个goroutine来处理与某个用户相关的消息通道。在处理消息之后，用户状态由goroutine更新并存储在数据库中。当对数据库的请求正在进行中时，可以向信道发送许多消息。在向数据库发送另一个请求之前，我想对它们进行处理。

目前，我正在使用len(ch)检查通道中的消息数量，并在for循环中读取它们。

func (c *consumer) handleUser(userID string, ch chan Message) {
  user := c.db.LoadUser(userID)
  for {
    var msgs []Message
    for n := len(ch); n > 0; n-- {
      msgs = append(msgs, <-ch)
    }
    apply.Messages(user, msgs)
    c.db.SaveUser(user)
}

ch := make(chan Message, 100)
go c.handleUser("user-1", ch) 

我在互联网上搜索，如果这是一些常见的模式，但我找不到类似的解决方案，我想知道我的方法是否有效/惯用的围棋程序。

Answer 1

您的解决方案将导致生成的goroutine在通道上旋转，直到至少发送一条消息。换句话说，戈鲁蒂河从来不阻塞。

在这里，您试图在一个批处理多个消息。有不同的方法来实现这一点。但是要回答的主要问题是:您如何知道这批消息是完整的？发送方可能具有此知识，并且可以将所有消息打包到一个片段中。另一方面，您可能不知道批处理何时准备好。在这些情况下，您需要使用超时，如下面的示例所示。

func (c *consumer) handleUser(userID string, ch chan Message) {
    user := c.db.LoadUser(userID)
    for {
        var msgs []Message
        select {
        case msg := <-ch:
            //Append the message in the current batch slice
            msgs = append(msgs, msg)
        //Wait up to 5 seconds and then process the batch
        case <-time.After(time.Second * 5):
            //Timeout: process the batch of messages
            if len(msgs) > 0 {
                apply.Messages(user, msgs)
                c.db.SaveUser(user)
            }
        }
    }
}

请注意，执行此函数的可能的goroutine只在实际有事情可做时才运行。