Go语言中实现带超时机制的批量消息处理

package main

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

// Message 定义了我们要处理的消息类型
type Message int

const (
    // CacheLimit 定义了消息缓存的最大容量
    CacheLimit = 100
    // CacheTimeout 定义了消息缓存的超时时间
    CacheTimeout = 5 * time.Second
)

func main() {
    // input 是一个带缓冲的通道，用于接收消息
    input := make(chan Message, CacheLimit)

    // 启动 poll Goroutine 来处理消息
    go poll(input)
    // 启动 generate Goroutine 来模拟生成消息
    generate(input)
}

// poll 检查传入消息并将其内部缓存，直到达到最大数量或超时。
func poll(input <-chan Message) {
    // cache 用于存储待发送的消息
    cache := make([]Message, 0, CacheLimit)
    // tick 是一个定时器，用于触发超时事件
    tick := time.NewTicker(CacheTimeout)

    for {
        select {
        // 情况1：有新消息到达
        case m := <-input:
            cache = append(cache, m) // 将消息添加到缓存

            // 如果缓存未达到上限，则继续等待
            if len(cache) < CacheLimit {
                break
            }

            // 如果缓存达到上限，则立即发送
            // 停止当前定时器，避免在发送后立即触发超时
            tick.Stop()

            // 发送缓存中的消息并清空缓存
            send(cache)
            cache = cache[:0] // 使用切片重切片清空，保留底层数组，提高效率

            // 重新创建一个定时器，以确保超时机制从此刻重新开始计时
            tick = time.NewTicker(CacheTimeout)

        // 情况2：定时器超时
        case <-tick.C:
            // 无论缓存大小，只要超时就发送
            send(cache)
            cache = cache[:0] // 清空缓存
        }
    }
}

// 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) {
    for {
        select {
        // 以随机时间间隔（0-100毫秒）生成消息
        case <-time.After(time.Duration(rand.Intn(100)) * time.Millisecond):
            input <- Message(rand.Int()) // 推送一个随机整数作为消息
        }
    }
}

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 检查传入消息并将其内部缓存，直到达到最大数量或超时。
func poll(input <-chan Message) {
    cache := make([]Message, 0, CacheLimit)
    tick := time.NewTicker(CacheTimeout)

    for {
        select {
        // 情况1：有新消息到达
        case m := <-input:
            cache = append(cache, m)

            if len(cache) < CacheLimit {
                break // 缓存未满，继续等待
            }

            // 缓存已满，立即发送
            tick.Stop() // 停止当前定时器
            send(cache)
            cache = cache[:0] // 清空缓存
            tick = time.NewTicker(CacheTimeout) // 重新创建定时器

        // 情况2：定时器超时
        case <-tick.C:
            send(cache)
            cache = cache[:0]
        }
    }
}

// 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) {
    for {
        select {
        case <-time.After(time.Duration(rand.Intn(100)) * time.Millisecond):
            input <- Message(rand.Int())
        }
    }
}