Go 并发原语 - Cond

// ❌ 轮询方案：浪费 CPU，响应慢
for len(queue) == maxSize {
    time.Sleep(10 * time.Millisecond) // 空转等待
}
queue = append(queue, item)

var mu sync.Mutex
cond := sync.NewCond(&mu)

package main

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

type BoundedQueue struct {
	cond    *sync.Cond
	queue   []int
	maxSize int
}

func NewBoundedQueue(maxSize int) *BoundedQueue {
	return &BoundedQueue{
		cond:    sync.NewCond(&sync.Mutex{}),
		maxSize: maxSize,
	}
}

func (q *BoundedQueue) Put(item int) {
	q.cond.L.Lock()
	defer q.cond.L.Unlock()

	for len(q.queue) == q.maxSize { // ✅ 必须用 for，不能用 if
		q.cond.Wait() // 队列满了，阻塞等待
	}

	q.queue = append(q.queue, item)
	fmt.Printf("放入: %d, 队列长度: %d\n", item, len(q.queue))
	q.cond.Broadcast() // 通知所有等待者（消费者可能在等）
}

func (q *BoundedQueue) Take() int {
	q.cond.L.Lock()
	defer q.cond.L.Unlock()

	for len(q.queue) == 0 { // ✅ 必须用 for，不能用 if
		q.cond.Wait() // 队列空了，阻塞等待
	}

	item := q.queue[0]
	q.queue = q.queue[1:]
	fmt.Printf("取出: %d, 队列长度: %d\n", item, len(q.queue))
	q.cond.Broadcast() // 通知所有等待者（生产者可能在等）
	return item
}

func main() {
	q := NewBoundedQueue(3) // 容量为 3

	// 启动 5 个生产者
	for i := 0; i < 5; i++ {
		go func(id int) {
			for j := 0; j < 3; j++ {
				q.Put(id*100 + j)
				time.Sleep(50 * time.Millisecond)
			}
		}(i)
	}

	// 启动 3 个消费者
	for i := 0; i < 3; i++ {
		go func() {
			for j := 0; j < 5; j++ {
				q.Take()
				time.Sleep(80 * time.Millisecond)
			}
		}()
	}

	time.Sleep(3 * time.Second)
}

// ❌ 用 if：可能在条件不满足时继续执行
cond.L.Lock()
if !condition {
    cond.Wait()
}
// 到这里条件不一定为 true！
doSomething()
cond.L.Unlock()

// ✅ 用 for：每次唤醒都重新检查条件
cond.L.Lock()
for !condition {
    cond.Wait()
}
// 到这里条件一定为 true
doSomething()
cond.L.Unlock()

// ❌ 没有 Lock 就 Wait
cond.Wait() // panic: sync: unlock of unlocked mutex

// ✅ 必须先获取锁
cond.L.Lock()
for !condition {
    cond.Wait()
}
cond.L.Unlock()

// ❌ 通知后没释放锁
cond.L.Lock()
condition = true
cond.Signal()
// 忘记 Unlock！被唤醒的 goroutine 卡在获取锁上 💀

// ✅ 通知后释放锁
cond.L.Lock()
condition = true
cond.Signal()
cond.L.Unlock() // 被唤醒的 goroutine 才能拿到锁

// ❌ 复制 Cond
func process(c sync.Cond) { // 值传递，复制了 Cond
    c.L.Lock()
    c.Wait()
    c.L.Unlock()
}

// ✅ 传指针
func process(c *sync.Cond) {
    c.L.Lock()
    c.Wait()
    c.L.Unlock()
}

Wait():
  1. 获取一个自增的 ticket 编号，加入等待链表
  2. 释放锁 (cond.L.Unlock())
  3. 信号量休眠，直到自己的 ticket 被通知
  4. 重新获取锁 (cond.L.Lock())

Signal():
  notify 计数 + 1
  唤醒等待链表中下一个未被通知的 goroutine

Broadcast():
  notify 计数直接设为 wait 计数
  唤醒等待链表中所有未被通知的 goroutine

// 方案 A：Cond（等待任意条件）
cond.L.Lock()
for len(queue) == 0 {
    cond.Wait()
}
item := queue[0]
queue = queue[1:]
cond.L.Unlock()

// 方案 B：Channel（天然的生产者-消费者）
ch := make(chan int, maxSize)
ch <- item   // 满了自动阻塞
item := <-ch // 空了自动阻塞

package main

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

type Service struct {
	cond  *sync.Cond
	ready bool
}

func NewService() *Service {
	return &Service{
		cond: sync.NewCond(&sync.Mutex{}),
	}
}

// Init 模拟耗时初始化
func (s *Service) Init() {
	fmt.Println("服务初始化中...")
	time.Sleep(2 * time.Second) // 模拟耗时操作

	s.cond.L.Lock()
	s.ready = true
	s.cond.Broadcast() // ✅ 唤醒所有等待的 worker
	s.cond.L.Unlock()
	fmt.Println("服务初始化完成")
}

// WaitReady 阻塞直到服务就绪
func (s *Service) WaitReady() {
	s.cond.L.Lock()
	for !s.ready {
		s.cond.Wait()
	}
	s.cond.L.Unlock()
}

func main() {
	svc := NewService()

	// 启动 5 个 worker，都需要等服务就绪
	var wg sync.WaitGroup
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			fmt.Printf("Worker %d: 等待服务就绪...\n", id)
			svc.WaitReady()
			fmt.Printf("Worker %d: 开始工作\n", id)
		}(i)
	}

	// 启动初始化
	go svc.Init()

	wg.Wait()
	fmt.Println("所有 Worker 完成")
}

# 两个应该加入 CI 的命令（和前面的文章一样）
go vet ./...
go test -race ./...