Golangchannel实现广播与多消费者模式-Golang-PHP中文网

Golangchannel实现广播与多消费者模式

P粉602998670

发布： 2025-09-22 15:45:01

原创

915人浏览过

Go语言通过channel实现并发通信，支持广播（一对多）和多消费者（多对一）模式。广播模式需自定义结构体维护多个channel，发送时遍历所有接收者；多消费者模式利用单一channel由多个goroutine竞争消费，适用于任务分发。两者结合可构建事件驱动的复杂系统。

golangchannel实现广播与多消费者模式

在Go语言中，channel 是实现并发通信的核心机制。通过合理设计 channel 的使用方式，可以轻松实现广播（一对多）和多消费者（多对一）模式。这两种模式在消息系统、事件通知、任务分发等场景中非常常见。

广播模式：一个生产者，多个消费者

广播模式要求一个发送者的消息能被多个接收者同时收到。由于Go的channel默认是“一出一入”（一个值只能被一个goroutine接收），要实现广播需要额外设计。

常见做法是：复制channel或使用闭包将消息推送给多个监听者。

示例：手动向多个channel发送相同消息

定义一个广播结构体，维护一组channel，每次发送时遍历所有channel：

type Broadcaster struct {
    receivers []chan string
    add       chan chan string
    remove    chan chan string
    message   chan string
}
<p>func NewBroadcaster() *Broadcaster {
b := &Broadcaster{
receivers: make([]chan string, 0),
add:       make(chan chan string),
remove:    make(chan chan string),
message:   make(chan string),
}
go b.run()
return b
}</p><p>func (b *Broadcaster) run() {
for {
select {
case ch := <-b.add:
b.receivers = append(b.receivers, ch)
case ch := <-b.remove:
for i, receiver := range b.receivers {
if receiver == ch {
b.receivers = append(b.receivers[:i], b.receivers[i+1:]...)
break
}
}
case msg := <-b.message:
for _, ch := range b.receivers {
select {
case ch <- msg:
default:
}
}
}
}
}</p>

登录后复制

使用方式：

立即学习“go语言免费学习笔记（深入）”；

<pre class="brush:php;toolbar:false;">b := NewBroadcaster()
<p>// 启动两个消费者
ch1 := make(chan string)
ch2 := make(chan string)</p><p>go func() {
for msg := range ch1 {
fmt.Println("Consumer 1 received:", msg)
}
}()</p><p>go func() {
for msg := range ch2 {
fmt.Println("Consumer 2 received:", msg)
}
}()</p><p>// 注册消费者
b.add <- ch1
b.add <- ch2</p><p>// 发送广播消息
b.message <- "Hello everyone!"</p>

登录后复制

这种方式实现了真正的广播：所有注册的消费者都能收到消息。

多消费者模式：多个消费者处理同一队列

这是典型的“工作池”模型：一个生产者发送任务，多个消费者从中取任务处理，每个任务只被一个消费者处理。

利用channel的天然竞争机制，多个goroutine从同一个channel读取，Go runtime会自动保证数据不被重复消费。

播记

播客shownotes生成器 | 为播客创作者而生

查看详情

示例：任务分发系统

创建一个共享channel，启动多个worker从其中读取任务：

<pre class="brush:php;toolbar:false;">func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d processing job %d\n", id, job)
        time.Sleep(time.Second) // 模拟处理时间
        results <- job * 2
    }
}
<p>func main() {
jobs := make(chan int, 100)
results := make(chan int, 100)</p><pre class="brush:php;toolbar:false;"><pre class="brush:php;toolbar:false;">// 启动3个消费者
for w := 1; w <= 3; w++ {
    go worker(w, jobs, results)
}

// 发送5个任务
for j := 1; j <= 5; j++ {
    jobs <- j
}
close(jobs)

// 收集结果
for a := 1; a <= 5; a++ {
    <-results
}

登录后复制

}

运行后可以看到不同worker轮流处理任务，实现了负载均衡。