功能说明
这段代码实现一个进程内发布订阅广播器：多个订阅者各自拿到一个缓冲通道，发布者把事件扇出给所有订阅者。如果某个订阅者的缓冲区已满，事件会对该订阅者丢弃，但发布者和其他订阅者继续工作。
这种“慢消费者降级”策略适合实时通知、指标推送或 WebSocket 广播：单个卡住的客户端不应拖垮整个服务。返回的投递数量可以用于监控丢弃率。
关键代码
Broker 用 sync.RWMutex 保护订阅者映射：Publish 持有读锁，Subscribe 和 Unsubscribe 持有写锁。这样发布过程中不会有订阅者被删除或通道被关闭。
Publish 遍历所有通道，使用带 default 分支的 select 做非阻塞发送；发送成功才增加计数。Unsubscribe 在写锁内先删除再关闭通道，重复调用是安全的。
使用注意
通道关闭后，消费者仍可读完缓冲区中的事件，然后收到零值；消费者不应再向通道发送数据。发布者不会重试被丢弃的事件，需要可靠性时应引入持久队列或确认机制。
缓冲区大小决定能容忍的突发量，过大会增加内存占用，过小会提高丢弃率。Publish 持有读锁期间执行发送，虽然是非阻塞的，但订阅者很多时仍会增加延迟。不要在订阅者回调里调用会长时间阻塞的操作。
验证方式
示例中快订阅者缓冲区为 4，慢订阅者为 1。连续发布 a、b、c 时，投递数依次为 2、1、1，慢订阅者只保留 a。退订慢订阅者后再发布 d 只投递给 1 个订阅者，全部退订后发布 e 返回 0，代码对这些结果都有断言。
代码示例
package main
import (
"fmt"
"sync"
)
type Broker struct {
mu sync.RWMutex
subscribers map[int]chan string
nextID int
}
func NewBroker() *Broker {
return &Broker{subscribers: make(map[int]chan string)}
}
func (b *Broker) Subscribe(buffer int) (int, <-chan string) {
b.mu.Lock()
defer b.mu.Unlock()
id := b.nextID
b.nextID++
channel := make(chan string, buffer)
b.subscribers[id] = channel
return id, channel
}
func (b *Broker) Unsubscribe(id int) {
b.mu.Lock()
defer b.mu.Unlock()
if channel, ok := b.subscribers[id]; ok {
delete(b.subscribers, id)
close(channel)
}
}
func (b *Broker) Publish(event string) int {
b.mu.RLock()
defer b.mu.RUnlock()
delivered := 0
for _, channel := range b.subscribers {
select {
case channel <- event:
delivered++
default:
}
}
return delivered
}
func main() {
broker := NewBroker()
fastID, fast := broker.Subscribe(4)
defer broker.Unsubscribe(fastID)
slowID, slow := broker.Subscribe(1)
defer broker.Unsubscribe(slowID)
delivered := []int{broker.Publish("a"), broker.Publish("b"), broker.Publish("c")}
fmt.Println("delivered:", delivered)
if delivered[0] != 2 || delivered[1] != 1 || delivered[2] != 1 {
panic("unexpected delivery counts")
}
fastSeen := []string{<-fast, <-fast, <-fast}
slowSeen := []string{<-slow}
fmt.Println(fastSeen, slowSeen)
if slowSeen[0] != "a" {
panic("slow subscriber should keep only the first event")
}
broker.Unsubscribe(slowID)
if broker.Publish("d") != 1 {
panic("expected one subscriber after unsubscribe")
}
broker.Unsubscribe(fastID)
if broker.Publish("e") != 0 {
panic("expected no subscribers")
}
}