Go 标准库提供了 channel 来做协程间通信，但 channel 不支持「队列长度到达上限后单独统计丢弃次数」这类自定义策略；而在需要精细控制唤醒时机、需要复用同一把锁保护其他状态、或者需要实现「批量入队 / 批量出队」语义时，sync.Cond 是更合适的基础原语。本文从零实现一个泛型有界阻塞队列 BoundedQueue[T]，包含 Put（满时阻塞）、Offer（满时立即返回失败）、Get（空时阻塞）、Poll、Close 与优雅关闭语义，并逐个讲清 sync.Cond 的三个易错点：必须持有锁调用 Wait、必须用 for 而非 if 循环检查条件、必须避免丢失唤醒。
为什么不用 channel 直接堆一个队列
市面上常见的说法是「有界阻塞队列用带缓冲的 channel 就行」。这在一半场景下成立，剩下的一半会卡在以下几点：
- 无法区分「阻塞」与「非阻塞」入队。channel 用
select+default可以实现非阻塞发送，但无法在同一个类型上同时暴露「满了就等」和「满了就返回错误」两种 API 且共享同一份缓冲区。 - 无法批量搬迁。从队列一次性取走最多 N 个元素，用 channel 需要循环 N 次
select，中间可能被其他 goroutine 插入打断，语义不原子。 - 无法与外部状态共享锁。真实业务中队列长度往往要和「当前在途请求数」「限流令牌」一起原子更新，channel 的内部锁不可见，只能再加一把锁，出现锁嵌套。
- 关闭语义不同。channel 关闭后仍可读出残余元素，但向已关闭的 channel 发送会 panic，且
close不能被多方安全调用。自实现队列可以用布尔标志位定义「关闭后仍允许排空」的策略。
下面的实现基于 Go 1.18+（需要泛型）。所有代码可以直接放进 queue.go，配套的 queue_test.go 用标准 testing 包即可运行。
sync.Cond 的三个前置知识
在动手前必须把 sync.Cond 的行为钉死，否则写出来的队列大概率会在压力测试下偶发死锁。
Wait 必须持锁调用，并且会临时释放锁
cond.Wait() 的内部动作是：把当前 goroutine 加入等待列表 → 释放关联的 Locker → 挂起 → 被唤醒后重新获取 Locker → 返回。所以调用前必须已经 Lock()，否则会触发 sync: unlock of unlocked mutex 这类 panic（实际报错通常发生在解锁阶段，信息类似 fatal error: sync: unlock of unlocked mutex）。
条件判断必须放在 for 循环里
Signal 唤醒的 goroutine 不保证条件已经成立——可能被别的 goroutine 抢先消费掉了。标准库文档原话是「it is therefore the responsibility of the caller to check the condition in a loop」。写成 if len(q.items) == 0 { cond.Wait() } 会在高并发下出现「醒来发现队列还是空但已经继续往下走」的越界 panic。
Signal 与 Broadcast 的取舍
Signal 唤醒一个等待者，Broadcast 唤醒全部。有界队列里通常会写两个 Cond：notEmpty 给消费者等，notFull 给生产者等。一次 Put 只增加一个元素，理论上 notEmpty.Signal() 就够；但一旦支持批量操作或 Close，就必须用 Broadcast，否则可能出现「部分等待者永远等不到通知」。
经验法则：单元素变更用
Signal，状态发生「全局性」变化（关闭、批量清空、配置重置）用Broadcast。混合使用时，只要有任何一条路径会改变多个等待者关心的条件，就统一用Broadcast，代价只是几次无谓唤醒，换来的是不会丢唤醒。
完整实现
实现要点：用 sync.Mutex 保护 items 切片、closed 标志；用环形切片（简单做法是 append + copy 收缩）维护 FIFO；Put 在满时等待 notFull，Get 在空时等待 notEmpty。
package bqueue
import (
"errors"
"sync"
)
var (
ErrClosed = errors.New("bqueue: queue is closed")
ErrFull = errors.New("bqueue: queue is full")
)
// BoundedQueue 是一个泛型有界阻塞队列，零值不可用，请用 New 构造。
type BoundedQueue[T any] struct {
mu sync.Mutex
notEmpty *sync.Cond
notFull *sync.Cond
items []T
capacity int
closed bool
}
// New 创建一个容量为 capacity 的队列，capacity 必须大于 0。
func New[T any](capacity int) *BoundedQueue[T] {
if capacity <= 0 {
panic("bqueue: capacity must be positive")
}
q := &BoundedQueue[T]{
items: make([]T, 0, capacity),
capacity: capacity,
}
q.notEmpty = sync.NewCond(&q.mu)
q.notFull = sync.NewCond(&q.mu)
return q
}
// Len 返回当前元素个数，用于监控打点。
func (q *BoundedQueue[T]) Len() int {
q.mu.Lock()
defer q.mu.Unlock()
return len(q.items)
}
// Cap 返回容量。
func (q *BoundedQueue[T]) Cap() int { return q.capacity }
// Put 阻塞入队：队列满时等待，队列已关闭时返回 ErrClosed。
func (q *BoundedQueue[T]) Put(v T) error {
q.mu.Lock()
defer q.mu.Unlock()
for len(q.items) == q.capacity && !q.closed {
q.notFull.Wait()
}
if q.closed {
return ErrClosed
}
q.items = append(q.items, v)
q.notEmpty.Signal()
return nil
}
// Offer 非阻塞入队：队列满或已关闭时立即返回错误。
func (q *BoundedQueue[T]) Offer(v T) error {
q.mu.Lock()
defer q.mu.Unlock()
if q.closed {
return ErrClosed
}
if len(q.items) == q.capacity {
return ErrFull
}
q.items = append(q.items, v)
q.notEmpty.Signal()
return nil
}
// Get 阻塞出队：队列空时等待，队列已关闭且已排空时返回 ErrClosed。
func (q *BoundedQueue[T]) Get() (T, error) {
q.mu.Lock()
defer q.mu.Unlock()
var zero T
for len(q.items) == 0 && !q.closed {
q.notEmpty.Wait()
}
if len(q.items) == 0 {
// 只有 closed 且为空才会走到这里
return zero, ErrClosed
}
v := q.items[0]
q.items = q.items[1:]
q.notFull.Signal()
return v, nil
}
// Poll 非阻塞出队。
func (q *BoundedQueue[T]) Poll() (T, bool) {
q.mu.Lock()
defer q.mu.Unlock()
var zero T
if len(q.items) == 0 {
return zero, false
}
v := q.items[0]
q.items = q.items[1:]
q.notFull.Signal()
return v, true
}
// GetBatch 一次性取出最多 n 个元素，返回的切片长度可能小于 n。
func (q *BoundedQueue[T]) GetBatch(n int) []T {
if n <= 0 {
return nil
}
q.mu.Lock()
defer q.mu.Unlock()
for len(q.items) == 0 && !q.closed {
q.notEmpty.Wait()
}
if len(q.items) == 0 {
return nil
}
if n > len(q.items) {
n = len(q.items)
}
out := make([]T, n)
copy(out, q.items[:n])
q.items = q.items[n:]
q.notFull.Broadcast()
return out
}
// Close 关闭队列，唤醒所有等待者。关闭后 Put/Offer 返回 ErrClosed，
// Get 仍可继续排空剩余元素，排空后返回 ErrClosed。
func (q *BoundedQueue[T]) Close() {
q.mu.Lock()
defer q.mu.Unlock()
if q.closed {
return
}
q.closed = true
q.notEmpty.Broadcast()
q.notFull.Broadcast()
}
几个设计决策值得单独说明：
q.items = q.items[1:]的隐藏成本。这种写法不复制元素，但每次出队都会让底层数组的起始指针前移，而cap不变。长期运行后底层数组会「左移」，直到append触发一次扩容才回收。对于高频出入队的队列，应该换成真正的环形缓冲（用head、tail下标 + 取模），下面给出改进版。- 关闭后仍允许排空。这是为了让消费者能处理完残留数据再退出，配合
for { v, err := q.Get(); if errors.Is(err, ErrClosed) { return } ... }使用非常自然。 - 故意
defer q.mu.Unlock()而非手动解锁。虽然Wait会临时解锁，但由于Wait返回前会重新加锁，外部逻辑上始终持锁，defer写法安全且不易出错。
换成环形缓冲，避免底层数组滑动
把 items 换成固定长度数组加 head、size 两个下标，出队只移动下标，不做切片操作，内存占用恒定。这是生产环境应当使用的版本。
package bqueue
import (
"errors"
"sync"
)
var (
ErrClosed = errors.New("bqueue: queue is closed")
ErrFull = errors.New("bqueue: queue is full")
)
type BoundedQueue[T any] struct {
mu sync.Mutex
notEmpty *sync.Cond
notFull *sync.Cond
buf []T
head int
size int
capacity int
closed bool
}
func New[T any](capacity int) *BoundedQueue[T] {
if capacity <= 0 {
panic("bqueue: capacity must be positive")
}
q := &BoundedQueue[T]{
buf: make([]T, capacity),
capacity: capacity,
}
q.notEmpty = sync.NewCond(&q.mu)
q.notFull = sync.NewCond(&q.mu)
return q
}
func (q *BoundedQueue[T]) Len() int {
q.mu.Lock()
defer q.mu.Unlock()
return q.size
}
func (q *BoundedQueue[T]) Cap() int { return q.capacity }
func (q *BoundedQueue[T]) Put(v T) error {
q.mu.Lock()
defer q.mu.Unlock()
for q.size == q.capacity && !q.closed {
q.notFull.Wait()
}
if q.closed {
return ErrClosed
}
q.push(v)
q.notEmpty.Signal()
return nil
}
func (q *BoundedQueue[T]) Offer(v T) error {
q.mu.Lock()
defer q.mu.Unlock()
if q.closed {
return ErrClosed
}
if q.size == q.capacity {
return ErrFull
}
q.push(v)
q.notEmpty.Signal()
return nil
}
func (q *BoundedQueue[T]) Get() (T, error) {
q.mu.Lock()
defer q.mu.Unlock()
var zero T
for q.size == 0 && !q.closed {
q.notEmpty.Wait()
}
if q.size == 0 {
return zero, ErrClosed
}
v := q.pop()
q.notFull.Signal()
return v, nil
}
func (q *BoundedQueue[T]) Close() {
q.mu.Lock()
defer q.mu.Unlock()
if q.closed {
return
}
q.closed = true
q.notEmpty.Broadcast()
q.notFull.Broadcast()
}
// push 与 pop 必须在持有 q.mu 时调用。
func (q *BoundedQueue[T]) push(v T) {
tail := (q.head + q.size) % q.capacity
q.buf[tail] = v
q.size++
}
func (q *BoundedQueue[T]) pop() T {
v := q.buf[q.head]
var zero T
q.buf[q.head] = zero // 及时释放引用，防止内存泄漏
q.head = (q.head + 1) % q.capacity
q.size--
return v
}
pop 里的 q.buf[q.head] = zero 不能省。当 T 是指针或含指针的 struct 时，不清零会让已出队元素继续被底层数组引用，GC 无法回收，长跑服务的内存曲线会缓慢爬升——这是环形缓冲最经典的内存泄漏点。
正确性验证：三个必须覆盖的测试
并发队列只靠肉眼审查不够，下面三个测试分别覆盖阻塞语义、关闭语义和唤醒竞争。
测试 1：容量满时 Put 阻塞，消费后被唤醒
func TestPutBlocksWhenFull(t *testing.T) {
q := New[int](1)
if err := q.Put(1); err != nil {
t.Fatalf("first put: %v", err)
}
done := make(chan error, 1)
go func() { done <- q.Put(2) }()
select {
case err := <-done:
t.Fatalf("put should block, got %v", err)
case <-time.After(100 * time.Millisecond):
}
v, err := q.Get()
if err != nil || v != 1 {
t.Fatalf("get = %v, %v", v, err)
}
select {
case err := <-done:
if err != nil {
t.Fatalf("second put: %v", err)
}
case <-time.After(time.Second):
t.Fatal("put was not woken up")
}
}
测试 2：Close 唤醒所有阻塞的生产者
func TestCloseWakesBlockedPutters(t *testing.T) {
const producers = 8
q := New[int](1)
_ = q.Put(0) // 塞满，后续 Put 全部阻塞
errCh := make(chan error, producers)
for i := 0; i < producers; i++ {
go func(i int) { errCh <- q.Put(i + 1) }(i)
}
time.Sleep(50 * time.Millisecond) // 让生产者进入 Wait
q.Close()
for i := 0; i < producers; i++ {
select {
case err := <-errCh:
if !errors.Is(err, ErrClosed) {
t.Fatalf("expect ErrClosed, got %v", err)
}
case <-time.After(time.Second):
t.Fatal("blocked putter not released by Close")
}
}
}
测试 3：多生产者多消费者，总量守恒
func TestConcurrentSum(t *testing.T) {
const (
producers = 8
consumers = 8
perProducer = 1000
)
q := New[int](64)
var wg sync.WaitGroup
wg.Add(producers)
for p := 0; p < producers; p++ {
go func() {
defer wg.Done()
for i := 0; i < perProducer; i++ {
if err := q.Put(1); err != nil {
t.Errorf("put: %v", err)
return
}
}
}()
}
var got int64
wg.Add(consumers)
for c := 0; c < consumers; c++ {
go func() {
defer wg.Done()
for {
v, err := q.Get()
if errors.Is(err, ErrClosed) {
return
}
if err != nil {
t.Errorf("get: %v", err)
return
}
atomic.AddInt64(&got, int64(v))
}
}()
}
// 所有生产者退出后关闭，消费者排空后自然退出
go func() {
time.Sleep(100 * time.Millisecond) // 粗略等待，真实用例应换成 WaitGroup
q.Close()
}()
wg.Wait()
if want := int64(producers * perProducer); got != want {
t.Fatalf("sum = %d, want %d", got, want)
}
}
用 go test -race -run TestConcurrentSum -count=200 反复跑，可以复现绝大多数丢失唤醒与竞态问题。如果 -race 没有报 DATA RACE 但测试偶发失败，几乎一定是条件判断写成了 if。
容易踩的坑与压测报告
坑 1：在 Wait 之前没有重新检查条件
Wait 返回并不代表条件成立，只代表「有人调用了 Signal/Broadcast 且你被选中」。文档明确要求在循环中检查。用 if 的典型崩法：两个消费者同时被唤醒，第一个把唯一元素取走，第二个执行 q.items[0] 触发 panic: runtime error: index out of range [0] with length 0。
坑 2：Signal 与 Wait 之间的窗口不可怕，丢失唤醒才可怕
因为 Wait 的加入等待列表与释放锁是在同一把锁保护下完成的，只要调用方在持锁状态下检查条件并进入 Wait，就不存在「信号发在检查之后、等待之前」的丢失唤醒。真正的丢失唤醒来自两类代码：一是在锁外修改被保护状态；二是忘记在 Close 里 Broadcast。
坑 3：Close 未做幂等保护
Close 里的 if q.closed { return } 必须保留。虽然对 sync.Cond 重复 Broadcast 不会 panic，但业务上 Close 往往触发其他清理逻辑，幂等是调用方默认会依赖的契约。
坑 4：忘记测试 Close 之后再 Put
Put 的循环条件是 for q.size == q.capacity && !q.closed，若写成 for q.size == q.capacity，关闭时消费者已不再消费，等待 notFull 的生产者会被 Broadcast 唤醒后立刻重新进入 Wait，形成永久阻塞。这类 bug 只在「队列满 + 关闭」同时发生时才暴露，务必写测试。
坑 5：GetBatch 用 Signal 导致部分等待者饿死
批量出队一次腾出 N 个空位，如果只 Signal 一个生产者，其余生产者虽然有空位但仍在睡。批量路径一律用 Broadcast。
小结
sync.Cond 不是过时 API，它是把「条件等待」这一语义暴露给调用方的唯一标准库手段。有界阻塞队列的正确实现可以压缩成三句话：条件检查永远放在 for 里；每次状态变更后在持锁期间发信号；关闭、批量、清空这类全局变化一律 Broadcast。缓冲区本身优先用环形结构，并在出队时清零槽位。把上面两个版本的代码放进同一个包，用 -race 跑通三个测试，再接入压测，基本可以放心用于生产。
FAQ
Q1：编译时报 sync: unlock of unlocked mutex 是怎么回事？
说明 Wait 被调用时当前 goroutine 没有持有对应的 sync.Mutex。检查是否漏了 q.mu.Lock()，或者用了 sync.NewCond(&sync.Mutex{}) 传了一个临时变量而不是结构体字段——后者会导致 Cond 锁的和你 Lock 的不是同一个对象。
Q2：能不能用 chan struct{} 取代 sync.Cond？
可以，常见做法是「信号量 + mutex」。但每个等待者都要维护自己的 channel，Broadcast 需要遍历所有等待者发送，代码量明显上升。另外 sync.Cond 的唤醒顺序与等待队列在运行时层面是公平的，自己实现容易在唤醒顺序上出现偏差。
Q3：Put 阻塞时调用 Close，返回 ErrClosed 是否会把数据丢了？
会丢，这是「关闭即拒绝新数据」语义。若业务要求不丢，应改为「关闭后仍允许写入直到排空」，或在关闭前由调用方确保所有生产者已停止（用 WaitGroup 等）。本实现选择前一种，因为它是大多数场景（服务停止时丢弃新请求）的默认预期。
Q4：为什么 GetBatch 返回 nil 而不是空切片？
两者在 len 上等价，但 nil 切片在序列化与比较上更省事，且能明确表达「没有取到任何元素」。如果调用方需要 append，nil 切片同样可以正常 append，不会出问题。
Q5：这个队列能直接用于高并发吗？有没有性能数据？
单锁 + 两个 Cond 的实现在中等并发（几十个 goroutine）下开销与带缓冲 channel 同量级，差别主要在唤醒策略：channel 是 per-waiter 的队列，sync.Cond 是共享等待列表。超过数百个并发生产者/消费者且对延迟敏感时，应考虑分片（sharding）成多个小队列，或改用 lock-free 环形队列。本文的实现适合作为分片方案的单分片单元。