如何在 Golang 中实现生产者消费者模型_Golang 并发任务协同设计示例-Golang-PHP中文网

如何在 Golang 中实现生产者消费者模型_Golang 并发任务协同设计示例

P粉602998670

发布： 2025-11-17 08:58:02

原创

430人浏览过

Golang中通过goroutine和channel实现生产者消费者模型，利用缓冲channel提升吞吐量，close通知结束，range自动检测关闭；结合sync.WaitGroup协调消费者退出，context控制超时与取消，确保程序健壮性。

如何在 golang 中实现生产者消费者模型_golang 并发任务协同设计示例

在 Golang 中实现生产者消费者模型，核心是利用 goroutine 和 channel 实现并发任务的协同。这种模型广泛应用于需要解耦数据生成与处理的场景，比如消息队列、任务调度、日志收集等。

使用 channel 构建基础生产者消费者

Go 的 channel 是天然的线程安全队列，非常适合实现生产者消费者模式。生产者将任务发送到 channel，消费者从 channel 接收并处理。

以下是一个简单示例：

package main
<p>import (
"fmt"
"time"
)</p><p>func producer(ch chan<- int) {
for i := 0; i < 5; i++ {
ch <- i
fmt.Printf("生产者: 生成 %d\n", i)
time.Sleep(100 * time.Millisecond)
}
close(ch) // 关闭 channel 表示不再生产
}</p><p>func consumer(ch <-chan int, id int) {
for data := range ch { // 自动检测 channel 是否关闭
fmt.Printf("消费者 %d: 处理 %d\n", id, data)
time.Sleep(200 * time.Millisecond)
}
}</p><p>func main() {
ch := make(chan int, 3) // 带缓冲的 channel</p><pre class='brush:php;toolbar:false;'>go producer(ch)

for i := 1; i <= 2; i++ {
    go consumer(ch, i)
}

time.Sleep(2 * time.Second) // 等待执行完成

登录后复制

}

歌者PPT

歌者PPT，AI 写 PPT 永久免费

197

查看详情

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

说明：

使用带缓冲的 channel 可以提升吞吐量。
生产者通过 close(ch) 通知消费者数据已全部发送。
消费者使用 for-range 自动接收数据并在 channel 关闭后退出循环。

控制消费者数量与优雅退出

实际应用中，通常固定消费者数量，并确保所有消费者处理完数据后再退出程序。可通过 sync.WaitGroup 协调。

func consumerWithWait(ch <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for data := range ch {
        fmt.Printf("消费者处理: %d\n", data)
        time.Sleep(150 * time.Millisecond)
    }
}
<p>func main() {
ch := make(chan int, 5)
var wg sync.WaitGroup</p><pre class='brush:php;toolbar:false;'>go producer(ch)

// 启动 3 个消费者
for i := 0; i < 3; i++ {
    wg.Add(1)
    go consumerWithWait(ch, &wg)
}

wg.Wait() // 等待所有消费者完成
fmt.Println("所有任务处理完毕")

登录后复制

}

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

关键点：

每个消费者启动前调用 wg.Add(1)。
消费者函数结束时调用 wg.Done()。
主协程通过 wg.Wait() 阻塞直到所有消费者退出。

支持取消操作：使用 context 控制生命周期

当需要提前终止生产或消费过程（如超时或中断信号），应使用 context 进行统一管理。

func producerWithContext(ctx context.Context, ch chan<- int) {
    for i := 0; ; i++ {
        select {
        case ch <- i:
            fmt.Printf("生产: %d\n", i)
            time.Sleep(100 * time.Millisecond)
        case <-ctx.Done():
            fmt.Println("生产者收到退出信号")
            close(ch)
            return
        }
    }
}
<p>func main() {
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()</p><pre class='brush:php;toolbar:false;'>ch := make(chan int, 5)
var wg sync.WaitGroup

go producerWithContext(ctx, ch)

for i := 0; i < 2; i++ {
    wg.Add(1)
    go func(id int) {
        defer wg.Done()
        for data := range ch {
            fmt.Printf("消费者 %d 处理: %d\n", id, data)
            time.Sleep(200 * time.Millisecond)
        }
    }(i)
}

wg.Wait()
fmt.Println("程序退出")

登录后复制

}

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

优势：