本文介绍使用 Go 通道（channel）构建一个持久化消费者函数，使其能持续接收来自多个生产者调用（如函数 a）的数据，并在累积满 N 个后统一处理，兼顾并发安全与资源可控性。
本文介绍使用 go 通道（channel）构建一个持久化消费者函数，使其能持续接收来自多个生产者调用（如函数 a）的输入，并在累积满 n 个后统一处理，兼顾并发安全与资源可控性。
在 Go 并发编程中，当多个函数调用（如 a() 的多次执行）需向单一处理逻辑（b()）异步提交数据，并要求该逻辑“攒够 N 个再统一处理”，本质是一个典型的带批处理语义的生产者-消费者模型。直接让 b() 每次调用都新建 goroutine 或依赖 sync.WaitGroup 并不可取——前者导致 goroutine 泄漏，后者无法自然表达“等待 N 个输入”的语义。正确解法是：由 b() 长期驻留为一个 goroutine，通过 channel 持续接收输入，并在内部完成计数与批处理。
✅ 推荐实现：持久化消费者 + 可选缓冲通道
以下是一个健壮、可复用的实现模板（以批量大小 N = 5 为例）：
package main
import (
"fmt"
"time"
)
type SomeType struct {
ID int
Data string
}
// 生产者：可被任意次数调用，非阻塞（若使用缓冲通道）
func a(input *SomeType, ch chan<- *SomeType) {
// 实际业务逻辑（如校验、转换等）
fmt.Printf("[a] Sending item #%d\n", input.ID)
ch <- input // 发送至通道
}
// 消费者：长期运行，按批次处理
func b(ch <-chan *SomeType, batchSize int) {
for {
batch := make([]*SomeType, 0, batchSize)
// 等待收集满 batchSize 个元素
for len(batch) < batchSize {
select {
case item := <-ch:
batch = append(batch, item)
fmt.Printf("[b] Collected %d/%d items\n", len(batch), batchSize)
}
}
// ✅ 批处理逻辑（此处仅为示例）
fmt.Printf("[b] Processing batch of %d items...\n", len(batch))
for _, item := range batch {
// 模拟耗时处理
time.Sleep(100 * time.Millisecond)
fmt.Printf(" → Processed #%d: %s\n", item.ID, item.Data)
}
fmt.Println("[b] Batch processing completed.\n")
}
}
func main() {
const N = 5
ch := make(chan *SomeType, N) // 使用缓冲通道，避免生产者阻塞
// 启动持久化消费者
go b(ch, N)
// 模拟多次生产者调用（可来自不同 goroutine）
for i := 1; i <= 12; i++ { // 总共发送 12 个，触发 2 个完整批次 + 1 个不完整批次（见下文说明）
go func(id int) {
a(&SomeType{
ID: id,
Data: fmt.Sprintf("payload-%d", id),
}, ch)
}(i)
time.Sleep(200 * time.Millisecond) // 模拟调用间隔
}
// 主 goroutine 短暂等待，确保输出可见（实际项目中应有优雅退出机制）
time.Sleep(3 * time.Second)
}⚠️ 关键注意事项
缓冲通道 vs 非缓冲通道：
示例中make(chan *SomeType, N)创建了容量为N的缓冲通道。这允许最多N次a()调用无需等待b()即可返回，提升吞吐量与响应性。若改用make(chan *SomeType)（无缓冲），则每次a()调用会阻塞，直到b()执行—— 适用于严格同步或背压控制场景。b()的无限循环设计：for { ... }确保b()持续工作，处理完一批后立即开始下一批。这是“持久化”的核心。若需支持优雅关闭（如程序退出时清空剩余数据），可引入context.Context和done通道，此处为简洁性暂略。未满批次的处理（进阶）：
上述代码默认等待严格满batchSize才处理。若需支持“超时兜底”（如 5 秒内未凑满也强制处理）或“程序结束前处理剩余项”，可在select中加入time.After()分支或监听os.Signal，属于生产环境必备增强。内存与 Goroutine 安全：
batch切片在每次循环中重新创建，避免跨批次数据污染；所有通道操作天然满足 Go 内存模型的同步保证，无需额外加锁。
✅ 总结
通过将消费者 b() 设计为长期运行的 goroutine，并结合 Go 原生 channel 的阻塞/缓冲特性，即可优雅解决“多点输入、单点批量处理”的需求。该方案轻量、高效、符合 Go 的并发哲学——用通信共享内存，而非用共享内存通信。实际项目中，只需根据业务调整 batchSize、处理逻辑及退出策略，即可快速落地。
