
在 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 incoming {
// 启动 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 并发哲学中“通过通信共享内存”的正确实践。










