17370845950

在 go 中，一个 channel 无法被多个 goroutine 同时“接收”同一消息；默认行为是竞争式消费。要实现“一个事件通知所有监听者”，需通过 fan-out 模式手动广播——即从源 channel 读取一次，再分别写入多个目标 channel。

Go 的 channel 是点对点通信原语，不具备内置广播能力。当你将同一个 incoming channel 同时传给 processEmail 和 processPagerDuty，两个 goroutine 实际上在竞争接收——每次仅有一个能成功读到事件，这正是你观察到“只有第一个 goroutine 收到事件”的根本原因。

要实现真正的“一对多”事件分发（即每个监听者都收到同一份事件副本），必须引入显式广播逻辑。推荐采用经典的 fan-out 模式：由一个中央分发 goroutine 从源 channel 读取事件，然后并发地、独立地将该事件发送至多个专用 consumer channel。以下是改造后的完整、可运行示例：

package main

import (
    "fmt"
    "time"
)

type Event struct {
    Host    string
    Command string
    Output  string
}

// 全局事件源（只供写入）
var incoming = make(chan Event, 10)

// 各服务专属接收 channel（缓冲避免阻塞分发器）
var (
    emailChan     = make(chan Event, 10)
    pagerDutyChan = make(chan Event, 10)
)

// 【关键】广播分发器：读取一次，发给所有订阅者
func broadcast() {
    for e := range incoming {
        // 并发发送，确保各 consumer 独立接收（不相互阻塞）
        go func(event Event) {
            select {
            case emailChan <- event:
            default:
                fmt.Println("⚠️  emailChan full, dropped event")
            }
        }(e)

        go func(event Event) {
            select {
            case pagerDutyChan <- event:
            default:
                fmt.Println("⚠️  pagerDutyChan full, dropped event")
            }
        }(e)
    }
}

func processEmail(ticker *time.Ticker) {
    for {
        select {
        case t := <-ticker.C:
            fmt.Println("? Email Tick at", t)
        case e := <-emailChan:
            fmt.Println("? EMAIL GOT AN EVENT!")
            fmt.Printf("%+v\n", e)
        }
    }
}

func processPagerDuty(ticker *time.Ticker) {
    for {
        select {
        case t := <-ticker.C:
            fmt.
Println("? PagerDuty Tick at", t)
        case e := <-pagerDutyChan:
            fmt.Println("? PAGERDUTY GOT AN EVENT!")
            fmt.Printf("%+v\n", e)
        }
    }
}

func eventAdd() {
    e := Event{
        Host:    "web01-east.domain.com",
        Command: "foo",
        Output:  "bar",
    }
    incoming <- e // 写入源 channel，触发广播
}

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

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

    pdTicker := time.NewTicker(1 * time.Second)
    go processPagerDuty(pdTicker)

    // 模拟 API 调用
    time.AfterFunc(2*time.Second, eventAdd)
    time.AfterFunc(5*time.Second, eventAdd)

    // 保持主 goroutine 运行
    select {}
}

✅ 关键设计要点说明：

• 分离关注点：incoming 是唯一输入入口；emailChan/pagerDutyChan 是各自逻辑的私有输入，解耦清晰。
• 非阻塞发送：使用 select { case ch
• 缓冲 channel：所有 channel 均设缓冲（如 make(chan T, 10)），防止瞬时高峰导致发送方阻塞或事件丢失。
• goroutine 安全：每个 go func(event Event){...}(e) 捕获当前事件值，避免循环变量闭包陷阱。

⚠️ 注意事项：

• 切勿在 broadcast() 中直接同步写入多个 channel（如 emailChan
• 若 consumer 可能长期阻塞或崩溃，建议增加健康检查与 channel 重连机制，或改用更健壮的消息中间件（如 NATS、Redis Pub/Sub）。
• 对于高吞吐场景，可考虑使用 sync.Pool 复用 Event 结构体指针，减少 GC 压力。

通过此模式，你既能保持 Go channel 的简洁性，又能精准实现事件广播语义——每个监听者都获得完整、独立的事件副本，真正达成“一个事件，多方响应”。