Go 中实现单通道多监听器的广播模式（Fan-Out）_技术教程

在 go 中，一个 channel 无法被多个 goroutine 同时“接收”同一份数据——默认行为是竞争式消费，仅有一个接收方能拿到消息。要实现“一个事件通知多个处理者”，需借助 fan-out 模式，通过 goroutine 复制消息并分发到多个独立 channel。

Go 的 channel 是点对点通信原语，不具备内置广播能力。当你将同一个 incoming chan Event 同时用于 processEmail 和 processPagerDuty 的随机选择一个就绪的接收方完成接收（基于调度公平性），因此你观察到“只有第一个启动的 goroutine 收到事件”——这并非 bug，而是 channel 的设计本质。

✅ 正确解法：使用 Fan-Out（扇出）模式 —— 由一个中央分发 goroutine 从源 channel 读取事件，并主动复制、并发发送给多个专用 listener channel：

type Event struct {
    Host, Command, Output string
}

// 定义多个独立的 listener channel（每个处理逻辑独占）
var (
    emailCh      = make(chan Event, 10)   // 缓冲避免阻塞分发者
    pagerDutyCh  = make(chan Event, 10)
    incoming     = make(chan Event, 10)   // 原始输入通道（如 HTTP API 写入）
)

// 中央分发器：读取 incoming，广播到所有 listener channel
func broadcast() {
    for e := range incoming {
        // 并发发送（非阻塞关键！需配合缓冲或 select 超时）
        go func(event Event) {
            select {
            case emailCh <- event:
            default: // 队列满时丢弃或记录告警（按业务需求调整）
                log.Println("emailCh full, dropped event")
            }
        }(e)

        go func(event Event) {
            select {
            case pagerDutyCh <- event:
            default:
                log.Println("pagerDutyCh full, dropped event")
            }
        }(e)
    }
}

// 各处理器保持原有结构，但监听专属 channel
func processEmail(ticker *time.Ticker) {
    for {
        select {
        case t := <-ticker.C:
            log.Println("Email Tick at", t)
        case e := <-emailCh: // ✅ 改为监听 emailCh
            log.Println("EMAIL GOT AN EVENT!", e)
        }
    }
}

func processPagerDuty(ticker *time.Ticker) {
    for {
        select {
        case t := <-ticker.C:
            log.Println("PagerDuty Tick at", t)
        case e := <-pagerDutyCh: // ✅ 改为监听 pagerDutyCh
            log.Println("PAGERDUTY GOT AN EVENT!", e)
        }
    }
}

func main() {
    // 启动广播器（必须在任何写入 incoming 前运行）
    go broadcast()

    emailTicker := time.NewTicker(10 * time.Second)
    pagerTicker := time.NewTicker(1 * time.Second)

    go processEmail(emailTicker)
    go processPagerDuty(pagerTicker)

    // 示例：模拟 API 事件注入
    go func() {
        http.HandleFunc("/event", func(w http.ResponseWriter, r *http.Request) {
            e := Event{
                Host: "web01-east.domain.com",
                Command: "check_disk",
                Output: "used: 87%",
            }
            incoming <- e // ✅ 写入统一入口
            w.WriteHeader(http.StatusOK)
        })
        log.Fatal(http.ListenAndServe(":8080", nil))
    }()

    select {} // 防止主 goroutine 退出
}

⚠️ 关键注意事项：

缓冲至关重要：所有 listener channel（emailCh, pagerDutyCh）必须设置合理缓冲容量（如示例中的 10），否则当某个处理器卡住（如网络延迟
、panic 未恢复），广播 goroutine 会在 select 的 case 中永久阻塞，导致整个事件流中断。
避免 goroutine 泄漏：示例中使用 select { case ch
不要直接关闭 incoming 后再 range：broadcast() 中的 for e := range incoming 依赖 channel 关闭，但生产环境通常长期运行，应通过 context 控制生命周期。
扩展性提示：若 listener 数量动态变化，可将 emailCh/pagerDutyCh 抽象为 []chan Event 切片，用 for _, ch := range listeners 循环分发。