Go批量读取channel的数据

package main

import (
        "fmt"
        "time"
)

func batchProcessor(ch <-chan string, batchSize int, flushInterval time.Duration) {
        var batch []string
        timer := time.NewTimer(flushInterval)

        for {
                select {
                case data := <-ch:
                        batch = append(batch, data)
                        // 当缓冲区达到批量大小时处理
                        if len(batch) >= batchSize {
                                fmt.Printf("Processing batch: %v\n", batch)
                                batch = nil
                                // 重置定时器
                                timer.Reset(flushInterval)
                        }

                case <-timer.C:
                        // 如果达到时间间隔，但 batch 不为空，也进行处理
                        if len(batch) > 0 {
                                fmt.Printf("Processing batch on timer: %v\n", batch)
                                batch = nil
                        }
                        // 重置定时器
                        timer.Reset(flushInterval)
                }
        }
}

func main() {
        dataChannel := make(chan string)
        batchSize := 5
        flushInterval := 3 * time.Second

        // 启动批量处理协程
        go batchProcessor(dataChannel, batchSize, flushInterval)

        // 模拟向 channel 发送数据
        for i := 1; i <= 10; i++ {
                dataChannel <- fmt.Sprintf("data-%d", i)
                time.Sleep(1 * time.Second)
        }

        // 让主程序暂停一会，以便查看处理结果
        time.Sleep(5 * time.Second)
}