在 Go 中，一个 channel 无法被多个 goroutine 同时“接收”同一消息；默认行为是竞争式消费。要实现“一个事件通知所有监听者”，需通过 fan-out 模式手动广播——即从源 channel 读取一次，再分别写入多个目标 channel。
Go 的 channel 是点对点通信原语，不具备内置广播能力。当你将同一个 incoming channel 同时传给 processEmail 和 processPagerDuty，两个 goroutine 实际上在竞争接收——每次仅有一个能成功读到事件，这正是你观察到“只有第一个 goroutine 收到事件”的根本原因。
要实现真正的“一对多”事件分发（即每个监听者都收到同一份事件副本），必须引入显式广播逻辑。推荐采用经典的 fan-out 模式：由一个中央分发 goroutine 从源 channel 读取事件，然后并发地、独立地将该事件发送至多个专用 consumer channel。以下是改造后的完整、可运行示例：
Miller CSV TSV JSON 数据处理器
下载
Miller (mlr) 是一个命令行工具，用于查询、整形和重新格式化名称索引数据，如 CSV、TSV、JSON 和 JSON Lines。它将 awk、sed、cut、join 和 sort 的功能整合到一个专为结构化数据处理而构建的单一工具中。
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 <- e: default: } 避免因某个 consumer 处理慢而拖垮整个广播流程（可根据业务需求替换为带超时的 select）。
- 缓冲 channel：所有 channel 均设缓冲（如 make(chan T, 10)），防止瞬时高峰导致发送方阻塞或事件丢失。
- goroutine 安全：每个 go func(event Event){...}(e) 捕获当前事件值，避免循环变量闭包陷阱。
⚠️ 注意事项：
- 切勿在 broadcast() 中直接同步写入多个 channel（如 emailChan <- e; pagerDutyChan <- e），否则任一 channel 阻塞都会卡住整个分发流程。
- 若 consumer 可能长期阻塞或崩溃，建议增加健康检查与 channel 重连机制，或改用更健壮的消息中间件（如 NATS、Redis Pub/Sub）。
- 对于高吞吐场景，可考虑使用 sync.Pool 复用 Event 结构体指针，减少 GC 压力。
通过此模式，你既能保持 Go channel 的简洁性，又能精准实现事件广播语义——每个监听者都获得完整、独立的事件副本，真正达成“一个事件，多方响应”。
