4. 带缓冲 channel 的生产者-消费者模型

带缓冲 channel 的生产者-消费者模型

带缓冲 channel 的核心价值在于解耦生产与消费的节奏：生产者不必等待消费者就绪即可连续投递，消费者也不必与生产者严格同步。缓冲容量 cap 决定了系统能容忍多大的瞬时速率差。当缓冲区写满时，生产者的 ch <- task 会阻塞，直到消费者取走一个元素腾出空位。

下面实现一个可观察的模型：生产者每 1 秒生成一个任务，消费者以可调的处理耗时并发取出处理。通过打印时间戳与缓冲区长度，可以直观看到缓冲填满、生产者阻塞、以及消费者加速后吞吐量回升的全过程。

package main

import (
	"fmt"
	"sync"
	"time"
)

type Task struct {
	ID int
}

func main() {
	const (
		bufSize      = 3               // 缓冲区容量
		taskCount    = 8               // 总任务数
		produceEvery = 1 * time.Second // 生产者节奏
		consumeCost  = 2 * time.Second // 单个消费者处理耗时
		consumers    = 2               // 消费者数量
	)

	ch := make(chan Task, bufSize)
	var wg sync.WaitGroup

	// 生产者：每秒投递一个任务
	wg.Add(1)
	go func() {
		defer wg.Done()
		defer close(ch)
		for i := 1; i <= taskCount; i++ {
			t := Task{ID: i}
			start := time.Now()
			ch <- t // 缓冲满时在此阻塞
			wait := time.Since(start)
			fmt.Printf("[%s] 生产 Task-%d  len=%d/%d  阻塞=%v\n",
				time.Now().Format("15:04:05"), t.ID, len(ch), cap(ch), wait.Round(time.Millisecond))
			time.Sleep(produceEvery)
		}
	}()

	// 消费者：并发取出并处理
	for c := 1; c <= consumers; c++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for t := range ch {
				fmt.Printf("[%s] 消费者%d 取出 Task-%d  len=%d/%d\n",
					time.Now().Format("15:04:05"), id, t.ID, len(ch), cap(ch))
				time.Sleep(consumeCost)
			}
		}(c)
	}

	wg.Wait()
	fmt.Println("全部任务处理完成")
}

运行后重点观察三件事：

• 缓冲区填充阶段：前 3 个任务几乎瞬间写入，len 从 0 涨到 3，生产者无阻塞。
• 阻塞出现：第 4 个任务开始，len=3/3 已满，生产者打印的 阻塞 值不再为 0，说明它被挂起等待消费者腾位。
• 吞吐量变化：2 个消费者并行处理，单个耗时 2 秒，理论消费速率约 1 个/秒，与生产速率持平；若把 consumeCost 调大或 consumers 调小，阻塞时间会显著增长。

关键参数与观察指标

通过修改上述参数并观察 len 与阻塞时间的变化，可以建立对带缓冲 channel 容量规划的基本直觉：缓冲不是越大越好，它只是吸收速率波动的弹性空间，真正的吞吐上限由消费端决定。