功能说明
生产速度高于消费速度时，无界缓冲会持续吃内存，最终拖垮进程。有界通道把压力反向传给生产者：缓冲满时发送方阻塞，消费一端腾出位置后才继续。示例用容量为二的通道、十个任务和每项二十毫秒的消费延迟，演示这种自然的限速效果。
流水线还会在截止时间到达时取消：生产者在发送处监听上下文，消费者在等待数据处监听上下文，两边都能及时退出，并由生产者负责关闭通道，避免向已关闭通道发送数据。
关键代码
produce 在 select 中同时等待发送成功和 ctx.Done，发送不出去时立即返回错误。consume 读取通道，通道关闭返回已收集结果；超时则返回 ctx.Err。runPipeline 创建带超时的上下文、缓冲通道和 WaitGroup；生产者协程用 defer 保证关闭通道，自身退出后消费者读到关闭信号自然结束。最后 wg.Wait 保证协程清理完成再返回。
使用注意
通道必须由写入方关闭，且只能关闭一次；本例中只有生产者关闭，消费者只读。超时后仍有未消费的数据留在缓冲里，会被垃圾回收，不会造成泄漏。消费函数里的 time.Sleep 只是模拟慢处理，真实代码应接收上下文以便尽快退出。生产者错误在超时场景属于预期，因此被过滤掉。容量过小会让调度开销变大，过大则失去背压意义。
验证方式
第一个流水线在五秒预算内处理全部十项，耗时约二百毫秒，err 为 nil。第二个流水线只给六十毫秒预算，只能消费约三项，deadline 打印为 true，说明上下文取消同时终止了两端。可以调整缓冲容量为一或五，观察耗时基本不变但生产者的阻塞频率不同。
代码示例
package main
import (
"context"
"errors"
"fmt"
"sync"
"time"
)
func produce(ctx context.Context, out chan<- int, total int) error {
for i := 1; i <= total; i++ {
select {
case out <- i:
case <-ctx.Done():
return ctx.Err()
}
}
return nil
}
func consume(ctx context.Context, in <-chan int, delay time.Duration) ([]int, error) {
var seen []int
for {
select {
case value, ok := <-in:
if !ok {
return seen, nil
}
time.Sleep(delay)
seen = append(seen, value)
case <-ctx.Done():
return seen, ctx.Err()
}
}
}
func runPipeline(total, buffer int, delay, budget time.Duration) ([]int, error) {
ctx, cancel := context.WithTimeout(context.Background(), budget)
defer cancel()
queue := make(chan int, buffer)
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
defer close(queue)
if err := produce(ctx, queue, total); err != nil && !errors.Is(err, context.DeadlineExceeded) {
fmt.Println("producer stopped:", err)
}
}()
seen, err := consume(ctx, queue, delay)
wg.Wait()
return seen, err
}
func main() {
start := time.Now()
seen, err := runPipeline(10, 2, 20*time.Millisecond, 5*time.Second)
fmt.Printf("consumed %d items in %s (err=%v)\n", len(seen), time.Since(start).Round(10*time.Millisecond), err)
short, err := runPipeline(10, 2, 20*time.Millisecond, 60*time.Millisecond)
fmt.Printf("budget run consumed %d items, deadline=%v\n", len(short), errors.Is(err, context.DeadlineExceeded))
}