Go，我如何一次从一个 channel 中提取 X 条消息

Question

我有一个带有传入消息的 channel 和一个等待它的例程我处理这些消息并将它们发送到不同的服务器

如果准备好，我想一次处理 100 条消息，或者在说 5 秒后处理那里的内容，然后再等待

我如何在 Go 中做到这一点

Answer 1

用于从消息 channel 读取的例程应该定义一个缓存，其中存储传入的消息。然后，当缓存达到 100 条消息或 5 秒后，这些缓存的消息将批量发送到远程服务器。您使用计时器 channel 和 Go 的 select 语句来确定哪个先发生。

以下示例可以在 Go playground 上运行

package main

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

type Message int

const (
    CacheLimit   = 100
    CacheTimeout = 5 * time.Second
)

func main() {
    input := make(chan Message, CacheLimit)

    go poll(input)
    generate(input)
}

// poll checks for incoming messages and caches them internally
// until either a maximum amount is reached, or a timeout occurs.
func poll(input <-chan Message) {
    cache := make([]Message, 0, CacheLimit)
    tick := time.NewTicker(CacheTimeout)

    for {
        select {
        // Check if a new messages is available.
        // If so, store it and check if the cache
        // has exceeded its size limit.
        case m := <-input:
            cache = append(cache, m)

            if len(cache) < CacheLimit {
                break
            }

            // Reset the timeout ticker.
            // Otherwise we will get too many sends.
            tick.Stop()

            // Send the cached messages and reset the cache.
            send(cache)
            cache = cache[:0]

            // Recreate the ticker, so the timeout trigger
            // remains consistent.
            tick = time.NewTicker(CacheTimeout)

        // If the timeout is reached, send the
        // current message cache, regardless of
        // its size.
        case <-tick.C:
            send(cache)
            cache = cache[:0]
        }
    }
}

// send sends cached messages to a remote server.
func send(cache []Message) {
    if len(cache) == 0 {
        return // Nothing to do here.
    }

    fmt.Printf("%d message(s) pending\n", len(cache))
}

// generate creates some random messages and pushes them into the given channel.
//
// Not part of the solution. This just simulates whatever you use to create
// the messages by creating a new message at random time intervals.
func generate(input chan<- Message) {
    for {
        select {
        case <-time.After(time.Duration(rand.Intn(100)) * time.Millisecond):
            input <- Message(rand.Int())
        }
    }
}