17370845950

在 go 中，一个 channel 默认只能被一个 goroutine 接收，无法直接“广播”给多个监听者；要实现事件同时通知多个处理协程，需借助 fan-out 模式——通过中间 goroutine 将每个事件复制并分发到多个独立 consumer channel。

Go 的 channel 是点对点通信机制，不具备内置广播能力。当你将同一个 incoming channel 同时传给 processEmail 和 processPagerDuty 两个 goroutine 时，每次 其中一个就绪的接收者（调度器随机选择），因此你观察到“只有第一个启动的 goroutine 收到事件”——这完全符合 Go channel 的语义，并非 bug，而是设计使然。

要让多个处理器同时收到同一事件，必须显式实现“事件复制 + 并行分发”，即经典的 fan-out 模式。核心思路是：
✅ 使用一个中央分发 goroutine，从源 channel 读取事件；
✅ 对每个事件，并发写入多个专用 consumer channel；
✅ 每个业务处理器（如邮件、PagerDuty）独占监听自己的 channel。

以下是针对你原始代码的重构方案（精简、可运行、生产就绪）：

package main

import (
    "fmt"
    "time"
)

type Event struct {
    Host    string
    Command string
    Output  string
}

var incoming = make(chan Event, 10) // 建议加缓冲，防阻塞发送端

// 为每个处理器创建专属 channel
var (
    emailCh       = make(chan Event, 10)
    pagerDutyCh   = make(chan Event, 10)
)

// 【关键】广播分发器：将每个事件复制并并发推送给所有消费者
func broadcast() {
    for e := range i
ncoming {
        // 启动 goroutine 并发写入，避免任一 consumer 阻塞导致整体卡死
        go func(event Event) {
            select {
            case emailCh <- event:
            default: // 非阻塞写入，丢弃或记录告警（根据业务需求）
                fmt.Println("WARN: emailCh full, dropping event")
            }
        }(e)

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

// 处理器逻辑保持不变，仅切换监听 channel
func processEmail(ticker *time.Ticker) {
    for {
        select {
        case t := <-ticker.C:
            fmt.Println("Email Tick at", t)
        case e := <-emailCh: // ← 改为监听 emailCh
            fmt.Println("EMAIL GOT AN EVENT!")
            fmt.Println(e)
        }
    }
}

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

func main() {
    // 启动广播器（必须在任何发送前启动！）
    go broadcast()

    // 启动处理器
    emailTicker := time.NewTicker(10 * time.Second)
    pagerTicker := time.NewTicker(1 * time.Second)
    go processEmail(emailTicker)
    go processPagerDuty(pagerTicker)

    // 模拟 API 事件注入（测试用）
    go func() {
        time.Sleep(2 * time.Second)
        incoming <- Event{
            Host:    "web01-east.domain.com",
            Command: "foo",
            Output:  "bar",
        }
    }()

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

⚠️ 关键注意事项：

• 永远不要在 broadcast 中直接同步写入多个 channel（如 emailCh
• 使用 go func(...){...}() + select{case ...: default:} 实现非阻塞、并发、容错分发；
• 所有 consumer channel 务必设置缓冲区（如 make(chan Event, 10)），否则慢消费者会拖垮系统；
• 若需严格保证事件不丢失，应结合重试、持久化队列（如 Redis Stream / NATS）等外部组件，channel 仅适合内存级、低延迟场景；
• broadcast() goroutine 是单点，但其内部并发写入天然支持水平扩展（增加更多 consumer channel 即可）。

通过此模式，你既能复用原有处理器逻辑，又能真正实现“一个事件、多方响应”的解耦架构——这才是 Go 并发哲学中“通过通信共享内存”的正确实践。