深入理解 Go Channel 批量读取与实际应用

Channel 是 Go 语言并发编程的核心，掌握批量读取技巧能显著提升程序性能。

一、Channel 基础回顾

1.1 Channel 类型

// 无缓冲 Channel
ch1 := make(chan int)

// 有缓冲 Channel
ch2 := make(chan int, 100)

// 只读 Channel
func reader(ch <-chan int) {}

// 只写 Channel
func writer(ch chan<- int) {}

1.2 基本操作

// 发送
ch <- value

// 接收
value := <-ch

// 检查 Channel 是否关闭
value, ok := <-ch
if !ok {
    // Channel 已关闭
}

// range 遍历
for value := range ch {
    process(value)
}

二、批量读取模式

2.1 基础批量读取

// 批量读取 Channel 数据
func batchRead(ch <-chan int, batchSize int) []int {
    batch := make([]int, 0, batchSize)
    
    for i := 0; i < batchSize; i++ {
        select {
        case v, ok := <-ch:
            if !ok {
                return batch
            }
            batch = append(batch, v)
        default:
            // Channel 为空，立即返回
            return batch
        }
    }
    
    return batch
}

2.2 带超时的批量读取

func batchReadWithTimeout(ch <-chan int, batchSize int, timeout time.Duration) []int {
    batch := make([]int, 0, batchSize)
    timer := time.NewTimer(timeout)
    defer timer.Stop()
    
    for i := 0; i < batchSize; i++ {
        select {
        case v, ok := <-ch:
            if !ok {
                return batch
            }
            batch = append(batch, v)
            
        case <-timer.C:
            // 超时，返回已收集的数据
            return batch
        }
    }
    
    return batch
}

2.3 动态批量读取

func dynamicBatchRead(ch <-chan int, minSize, maxSize int, maxWait time.Duration) []int {
    batch := make([]int, 0, maxSize)
    start := time.Now()
    
    for len(batch) < maxSize {
        remaining := maxWait - time.Since(start)
        if remaining <= 0 && len(batch) >= minSize {
            break
        }
        
        select {
        case v, ok := <-ch:
            if !ok {
                return batch
            }
            batch = append(batch, v)
            
        case <-time.After(remaining):
            if len(batch) >= minSize {
                return batch
            }
        }
    }
    
    return batch
}

三、实际应用场景

3.1 数据库批量写入

type LogEntry struct {
    Timestamp time.Time
    Level     string
    Message   string
}

type LogBatcher struct {
    input    chan LogEntry
    db       *sql.DB
    batchSize int
    flushInterval time.Duration
}

func (lb *LogBatcher) Start() {
    go lb.process()
}

func (lb *LogBatcher) process() {
    ticker := time.NewTicker(lb.flushInterval)
    defer ticker.Stop()
    
    batch := make([]LogEntry, 0, lb.batchSize)
    
    for {
        select {
        case entry, ok := <-lb.input:
            if !ok {
                // Channel 关闭，刷新剩余数据
                if len(batch) > 0 {
                    lb.flush(batch)
                }
                return
            }
            
            batch = append(batch, entry)
            
            if len(batch) >= lb.batchSize {
                lb.flush(batch)
                batch = batch[:0]
            }
            
        case <-ticker.C:
            if len(batch) > 0 {
                lb.flush(batch)
                batch = batch[:0]
            }
        }
    }
}

func (lb *LogBatcher) flush(batch []LogEntry) {
    // 批量插入数据库
    tx, err := lb.db.Begin()
    if err != nil {
        log.Printf("Failed to begin transaction: %v", err)
        return
    }
    defer tx.Rollback()
    
    stmt, err := tx.Prepare("INSERT INTO logs (timestamp, level, message) VALUES (?, ?, ?)")
    if err != nil {
        log.Printf("Failed to prepare statement: %v", err)
        return
    }
    defer stmt.Close()
    
    for _, entry := range batch {
        if _, err := stmt.Exec(entry.Timestamp, entry.Level, entry.Message); err != nil {
            log.Printf("Failed to insert log: %v", err)
        }
    }
    
    if err := tx.Commit(); err != nil {
        log.Printf("Failed to commit transaction: %v", err)
    }
}

3.2 消息队列批量消费

type Message struct {
    ID      string
    Payload []byte
}

type Consumer struct {
    messages chan Message
    handler  func([]Message) error
    batchSize int
    workers   int
}

func (c *Consumer) Start() {
    for i := 0; i < c.workers; i++ {
        go c.worker(i)
    }
}

func (c *Consumer) worker(id int) {
    batch := make([]Message, 0, c.batchSize)
    
    for msg := range c.messages {
        batch = append(batch, msg)
        
        if len(batch) >= c.batchSize {
            if err := c.handler(batch); err != nil {
                log.Printf("Worker %d: failed to process batch: %v", id, err)
            }
            batch = batch[:0]
        }
    }
    
    // 处理剩余消息
    if len(batch) > 0 {
        if err := c.handler(batch); err != nil {
            log.Printf("Worker %d: failed to process final batch: %v", id, err)
        }
    }
}

3.3 流式数据处理

type DataProcessor struct {
    input  chan DataPoint
    output chan ProcessedData
    window time.Duration
}

func (dp *DataProcessor) ProcessStream() {
    ticker := time.NewTicker(dp.window)
    defer ticker.Stop()
    
    window := make([]DataPoint, 0)
    
    for {
        select {
        case data, ok := <-dp.input:
            if !ok {
                // 处理最后窗口
                if len(window) > 0 {
                    dp.output <- dp.compute(window)
                }
                close(dp.output)
                return
            }
            window = append(window, data)
            
        case <-ticker.C:
            if len(window) > 0 {
                dp.output <- dp.compute(window)
                window = window[:0]
            }
        }
    }
}

func (dp *DataProcessor) compute(window []DataPoint) ProcessedData {
    // 计算窗口统计信息
    var sum, min, max float64
    min = math.MaxFloat64
    max = -math.MaxFloat64
    
    for _, dp := range window {
        sum += dp.Value
        if dp.Value < min {
            min = dp.Value
        }
        if dp.Value > max {
            max = dp.Value
        }
    }
    
    return ProcessedData{
        Count: len(window),
        Sum:   sum,
        Avg:   sum / float64(len(window)),
        Min:   min,
        Max:   max,
    }
}

四、高级技巧

4.1 多 Channel 批量合并

func mergeChannels(channels ...<-chan int) <-chan []int {
    out := make(chan []int)
    var wg sync.WaitGroup
    
    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            
            batch := make([]int, 0, 100)
            for v := range c {
                batch = append(batch, v)
                if len(batch) >= 100 {
                    out <- batch
                    batch = batch[:0]
                }
            }
            
            if len(batch) > 0 {
                out <- batch
            }
        }(ch)
    }
    
    go func() {
        wg.Wait()
        close(out)
    }()
    
    return out
}

4.2 优先级 Channel

type PriorityItem struct {
    Priority int
    Data     interface{}
}

type PriorityBatcher struct {
    highPriority chan PriorityItem
    lowPriority  chan PriorityItem
    output       chan []PriorityItem
    batchSize    int
}

func (pb *PriorityBatcher) Start() {
    go pb.process()
}

func (pb *PriorityBatcher) process() {
    batch := make([]PriorityItem, 0, pb.batchSize)
    
    for {
        // 优先处理高优先级数据
        select {
        case item := <-pb.highPriority:
            batch = append(batch, item)
        default:
            // 高优先级为空，处理低优先级
            select {
            case item := <-pb.highPriority:
                batch = append(batch, item)
            case item := <-pb.lowPriority:
                batch = append(batch, item)
            }
        }
        
        if len(batch) >= pb.batchSize {
            pb.output <- batch
            batch = batch[:0]
        }
    }
}

4.3 背压控制

type BackpressureBatcher struct {
    input    chan Task
    output   chan []Task
    maxSize  int
    maxWait  time.Duration
}

func (bp *BackpressureBatcher) Start() {
    go bp.process()
}

func (bp *BackpressureBatcher) process() {
    batch := make([]Task, 0, bp.maxSize)
    timer := time.NewTimer(bp.maxWait)
    defer timer.Stop()
    
    for {
        select {
        case task, ok := <-bp.input:
            if !ok {
                if len(batch) > 0 {
                    bp.output <- batch
                }
                close(bp.output)
                return
            }
            
            // 检查背压
            if len(batch) >= bp.maxSize {
                // 阻塞等待消费者
                bp.output <- batch
                batch = make([]Task, 0, bp.maxSize)
                timer.Reset(bp.maxWait)
            }
            
            batch = append(batch, task)
            
        case <-timer.C:
            if len(batch) > 0 {
                bp.output <- batch
                batch = batch[:0]
            }
            timer.Reset(bp.maxWait)
        }
    }
}

五、性能优化

5.1 预分配内存

// ❌ 不推荐：动态扩容
func badBatchRead(ch <-chan int) []int {
    var batch []int  // 初始容量为 0
    for v := range ch {
        batch = append(batch, v)  // 频繁扩容
    }
    return batch
}

// ✅ 推荐：预分配容量
func goodBatchRead(ch <-chan int, batchSize int) []int {
    batch := make([]int, 0, batchSize)  // 预分配
    for i := 0; i < batchSize; i++ {
        select {
        case v := <-ch:
            batch = append(batch, v)
        default:
            return batch
        }
    }
    return batch
}

5.2 批量大小选择

// 根据场景选择最佳批量大小
func getOptimalBatchSize(scenario string) int {
    switch scenario {
    case "database_write":
        return 1000  // 数据库批量写入
    case "api_request":
        return 100   // API 请求合并
    case "log_processing":
        return 500   // 日志处理
    case "realtime":
        return 10    // 实时性要求高
    default:
        return 100
    }
}

六、总结

模式	适用场景	核心要点
基础批量读取	简单数据聚合	控制批量大小，处理 Channel 关闭
超时批量读取	实时性要求	平衡延迟和吞吐量
动态批量读取	流量波动大	自适应调整批量大小
背压控制	生产者快于消费者	防止内存溢出
优先级批量	多优先级数据	优先处理高优先级

掌握 Channel 批量读取技术，能让你的 Go 程序在处理流式数据时更加高效和可靠。