Golang中Broadcast 和Signal区别小结

type Cond struct {
    L Locker          // 关联的互斥锁
    notify  notifyList // 通知队列
    checker copyChecker // 防止复制检查
}

cond := sync.NewCond(&sync.Mutex{})

// 等待方
cond.L.Lock()
for !condition {
    cond.Wait() // 原子解锁并挂起
}
// 执行操作
cond.L.Unlock()

// 通知方
cond.L.Lock()
// 改变条件
cond.Signal() // 或 cond.Broadcast()
cond.L.Unlock()

// runtime/sema.go

// Signal实现
func notifyListNotifyOne(l *notifyList) {
    // 从等待队列头部取出一个goroutine
    s := l.head
    if s != nil {
        l.head = s.next
        if l.head == nil {
            l.tail = nil
        }
        // 唤醒该goroutine
        readyWithTime(s, 4)
    }
}

// Broadcast实现
func notifyListNotifyAll(l *notifyList) {
    // 取出整个等待队列
    s := l.head
    l.head = nil
    l.tail = nil

    // 逆序唤醒所有goroutine（避免优先级反转）
    var next *sudog
    for s != nil {
        next = s.next
        s.next = nil
        readyWithTime(s, 4)
        s = next
    }
}

type TaskDispatcher struct {
    cond  *sync.Cond
    tasks []Task
}

func (d *TaskDispatcher) AddTask(task Task) {
    d.cond.L.Lock()
    d.tasks = append(d.tasks, task)
    d.cond.Signal() // 只唤醒一个worker
    d.cond.L.Unlock()
}

func (d *TaskDispatcher) Worker(id int) {
    for {
        d.cond.L.Lock()
        for len(d.tasks) == 0 {
            d.cond.Wait()
        }
        task := d.tasks[0]
        d.tasks = d.tasks[1:]
        d.cond.L.Unlock()
        
        processTask(id, task)
    }
}

type ConfigManager struct {
    cond   *sync.Cond
    config atomic.Value // 存储当前配置
}

func (m *ConfigManager) UpdateConfig(newConfig Config) {
    m.cond.L.Lock()
    m.config.Store(newConfig)
    m.cond.Broadcast() // 通知所有监听者
    m.cond.L.Unlock()
}

func (m *ConfigManager) WatchConfig() {
    for {
        m.cond.L.Lock()
        current := m.config.Load().(Config)
        
        // 等待配置变更
        m.cond.Wait()
        newConfig := m.config.Load().(Config)
        
        if newConfig.Version != current.Version {
            applyNewConfig(newConfig)
        }
        m.cond.L.Unlock()
    }
}

func BenchmarkSignal(b *testing.B) {
    cond := sync.NewCond(&sync.Mutex{})
    var wg sync.WaitGroup
    
    // 准备100个等待goroutine
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func() {
            cond.L.Lock()
            cond.Wait()
            cond.L.Unlock()
            wg.Done()
        }()
    }
    
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        cond.Signal() // 每次唤醒一个
    }
    
    // 清理
    cond.Broadcast()
    wg.Wait()
}

func BenchmarkBroadcast(b *testing.B) {
    cond := sync.NewCond(&sync.Mutex{})
    var wg sync.WaitGroup
    
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            // 每个迭代创建100个等待者
            for i := 0; i < 100; i++ {
                wg.Add(1)
                go func() {
                    cond.L.Lock()
                    cond.Wait()
                    cond.L.Unlock()
                    wg.Done()
                }()
            }
            
            cond.Broadcast() // 唤醒所有
            wg.Wait()
        }
    })
}

func (q *TaskQueue) Notify(n int) {
    q.cond.L.Lock()
    defer q.cond.L.Unlock()
    
    // 根据任务数量精确唤醒
    for i := 0; i < min(n, len(q.waiters)); i++ {
        q.cond.Signal()
    }
}

// 错误示例：可能造成永久阻塞
cond.L.Lock()
if len(tasks) > 0 {
    cond.Signal() // 可能无等待者
}
cond.L.Unlock()

cond.L.Lock()
hasTasks := len(tasks) > 0
cond.L.Unlock()

if hasTasks {
    cond.Signal() // 在锁外通知更安全
}

type StatusNotifier struct {
    cond    *sync.Cond
    version int64 // 状态版本号
}

func (s *StatusNotifier) UpdateStatus() {
    s.cond.L.Lock()
    s.version++ // 版本更新
    s.cond.Broadcast()
    s.cond.L.Unlock()
}

func (s *StatusNotifier) WaitForChange(ver int64) int64 {
    s.cond.L.Lock()
    defer s.cond.L.Unlock()
    
    for s.version == ver {
        s.cond.Wait()
        // 可能被虚假唤醒，检查版本
    }
    return s.version
}

var (
    cond = sync.NewCond(&sync.Mutex{})
    resource int
)

func consumer() {
    cond.L.Lock()
    for resource == 0 {
        cond.Wait() // 等待资源
    }
    resource--
    cond.L.Unlock()
}

func producer() {
    cond.L.Lock()
    resource += 5
    cond.Signal() // 错误：只唤醒一个消费者
    cond.L.Unlock()
}

// 正确做法：根据资源数量唤醒
for i := 0; i < min(5, resource); i++ {
    cond.Signal()
}

func process() {
    for {
        // 高频状态检查
        cond.L.Lock()
        if !ready {
            cond.Wait()
        }
        cond.L.Unlock()
        
        // 处理工作...
    }
}

func update() {
    // 每毫秒触发更新
    for range time.Tick(time.Millisecond) {
        cond.Broadcast() // 每秒唤醒1000次
    }
}

// 使用条件变量+状态标记
func update() {
    for range time.Tick(time.Millisecond) {
        cond.L.Lock()
        statusUpdated = true
        cond.Broadcast()
        cond.L.Unlock()
    }
}

func process() {
    lastStatus := 0
    for {
        cond.L.Lock()
        for !statusUpdated {
            cond.Wait()
        }
        
        // 获取最新状态
        current := getStatus()
        if current == lastStatus {
            // 状态未实际变化，跳过处理
            statusUpdated = false
            cond.L.Unlock()
            continue
        }
        
        lastStatus = current
        statusUpdated = false
        cond.L.Unlock()
        
        // 处理状态变化...
    }
}

graph TD
    A[需要通知goroutine] --> B{变更性质}
    B -->|资源可用| C[有多少资源？]
    C -->|单个资源| D[使用Signal]
    C -->|多个资源| E[多次Signal或条件Broadcast]
    B -->|状态变更| F[所有等待者都需要知道？]
    F -->|是| G[使用Broadcast]
    F -->|否| H[按需使用Signal]
    A --> I{性能要求}
    I -->|高并发低延迟| J[优先Signal]
    I -->|吞吐量优先| K[评估Broadcast开销]

// 使用Broadcast前确认：
if 状态变化影响所有等待者 &&
   无性能瓶颈风险 &&
   避免惊群效应措施 {
   cond.Broadcast()
}

// 正确：循环检查条件
for !condition {
    cond.Wait()
}

// 危险：if检查可能虚假唤醒
if !condition {
    cond.Wait()
}

// 跟踪Wait调用
func (c *TracedCond) Wait() {
    start := time.Now()
    c.Cond.Wait()
    metrics.ObserveWaitDuration(time.Since(start))
}