Go语言中实现通道消息的批量处理与定时刷新策略

package main

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

// Message 定义消息类型，此处为简化示例使用 int
type Message int

const (
    // CacheLimit 定义缓存中消息的最大数量
    CacheLimit = 100
    // CacheTimeout 定义批量发送的超时时间
    CacheTimeout = 5 * time.Second
)

func main() {
    // 创建一个带缓冲的输入通道，缓冲大小等于缓存限制
    // 缓冲通道可以在消费者忙碌时，允许生产者继续发送一定数量的消息而不被阻塞
    input := make(chan Message, CacheLimit)

    // 启动一个协程来轮询和处理消息（消费者）
    go poll(input)
    // 启动一个协程来生成消息（模拟生产者）
    generate(input)

    // 主协程通过一个空的select语句阻塞，防止程序立即退出。
    // 在实际应用中，通常会有更复杂的同步机制，例如等待所有协程完成。
    select {}
}

// poll 协程负责从输入通道接收消息，并将其缓存，直到达到数量限制或超时
func poll(input <-chan Message) {
    // 初始化消息缓存，预分配容量以减少扩容开销
    cache := make([]Message, 0, CacheLimit)
    // 创建一个周期性定时器，用于触发超时批量发送
    tick := time.NewTicker(CacheTimeout)
    // 使用 defer 确保在协程退出时停止定时器，释放资源
    defer tick.Stop()

    for {
        select {
        // case 1: 尝试从输入通道接收新消息
        case m := <-input:
            cache = append(cache, m) // 将消息添加到缓存

            // 如果缓存中的消息数量未达到预设的限制，则继续等待
            // 等待下一个消息或定时器触发
            if len(cache) < CacheLimit {
                break // 跳出 select，继续 for 循环等待
            }

            // 缓存已达到数量限制，立即进行批量发送
            // 停止当前定时器，避免在批量发送后，原定时器又立即触发一次超时发送
            tick.Stop()

            send(cache)      // 发送缓存中的消息
            cache = cache[:0] // 清空缓存，准备接收新的批次

            // 重新创建一个定时器，确保新的超时周期从现在开始计算
            // 这样，即使消息源源不断，每次达到数量上限发送后，都会有一个完整的超时周期
            tick = time.NewTicker(CacheTimeout)

        // case 2: 如果定时器触发，表示达到超时时间
        case <-tick.C:
            // 无论缓存中有多少消息（可能为0），都进行发送
            send(cache)
            cache = cache[:0] // 清空缓存

            // 注意：这里不需要重新创建定时器。
            // time.NewTicker 返回的通道在被读取后会自动重置，
            // 下一个周期会继续计时。只有在达到数量限制提前发送时才需要重置定时器。
        }
    }
}

// send 模拟将缓存中的消息发送到远程服务器或进行其他批量处理操作
func send(cache []Message) {
    if len(cache) == 0 {
        return // 缓存为空，无需操作
    }
    // 在实际应用中，这里会进行网络请求、数据库写入、文件写入等耗时操作
    fmt.Printf("[%s] 发送 %d 条消息\n", time.Now().Format("15:04:05"), len(cache))
}

// generate 模拟消息生成器，以随机间隔向通道推送消息
// 这部分代码仅用于模拟消息源，并非解决方案的核心组成部分
func generate(input chan<- Message) {
    msgID := 0
    for {
        select {
        // 随机等待一段短时间后生成一条新消息，模拟不均匀的消息流入
        case <-time.After(time.Duration(rand.Intn(100)) * time.Millisecond):
            msgID++
            input <- Message(msgID)
        }
    }
}