首页 > Golang > Go教程

Golangchannel实现生产者消费者模式

时间：2026-01-14 15:37:06 391浏览收藏

你在学习Golang相关的知识吗？本文《Golang channel实现生产者消费者模型》，主要介绍的内容就涉及到，如果你想提升自己的开发能力，就不要错过这篇文章，大家要知道编程理论基础和实战操作都是不可或缺的哦！

直接用 chan 做任务队列易出阻塞、死锁、任务丢失等问题，因其仅为通信原语，缺乏重试、ACK、积压监控等生产级能力；应结合 select/default、sync.WaitGroup 或封装 TaskQueue，必要时换用 Redis/Kafka。

如何使用Golang实现生产者消费者模型_Golang channel队列与任务管理实践

为什么直接用 `chan` 做任务队列容易出问题

Go 的 chan 天然适合生产者消费者模型，但直接裸用会导致阻塞、死锁或任务丢失。典型问题是：生产者往已满的无缓冲通道写入时永久阻塞；消费者 panic 后未关闭通道，导致其他 goroutine 无限等待；或者多个消费者竞争同一通道却没做任务确认机制，造成重复消费或漏消费。

关键点在于：chan 是通信原语，不是任务队列实现——它不提供重试、ACK、积压监控、优雅关闭等生产必需能力。

无缓冲通道（make(chan int)）要求收发双方同时就绪，否则阻塞
有缓冲通道（make(chan int, 100)）仅缓解压力，满后仍阻塞，且无法感知积压量
关闭通道后，接收方仍可读完剩余数据，但无法区分“空”和“已关闭”，易误判终止条件

用 `chan` + `sync.WaitGroup` 实现基础可靠模型

这是最轻量、可控性最强的实践方式，适用于中低频任务（如日志收集、配置同步），核心是让生产者不阻塞、消费者可退出、任务不丢失。

关键设计：

生产者用 select 配合 default 实现非阻塞写入，失败时走本地重试或丢弃策略
消费者用 range 遍历通道，配合 sync.WaitGroup 等待所有任务完成
关闭通道前确保所有生产者 goroutine 已退出，避免向已关闭通道写入 panic

func main() {
    tasks := make(chan string, 10)
    var wg sync.WaitGroup
<pre class="brush:php;toolbar:false;">// 启动 3 个消费者
for i := 0; i < 3; i++ {
    wg.Add(1)
    go func(id int) {
        defer wg.Done()
        for task := range tasks {
            fmt.Printf("worker %d processing: %s\n", id, task)
            time.Sleep(100 * time.Millisecond) // 模拟处理
        }
    }(i)
}

// 生产者：非阻塞写入
for i := 0; i < 20; i++ {
    select {
    case tasks <- fmt.Sprintf("task-%d", i):
    default:
        fmt.Printf("task-%d dropped: channel full\n", i)
    }
}

close(tasks)
wg.Wait()

}

加一层封装：带超时与错误反馈的 `TaskQueue`

当任务需要结果反馈、或必须保证至少一次交付时，裸通道就不够用了。此时应封装一个结构体，把输入通道、输出通道、错误通道、上下文控制统一管理。

注意点：

不要在消费者内部直接 panic，而是将错误发到 errCh，由主逻辑统一处理
用 context.Context 控制单个任务超时，避免某个慢任务拖垮整个队列
输出通道应为有缓冲（如 make(chan Result, len(tasks))），防止结果写入阻塞下一个任务处理

type Task struct {
    ID     string
    Data   []byte
    Ctx    context.Context
}
<p>type Result struct {
TaskID string
Err    error
Output interface{}
}</p><p>func NewTaskQueue(workers int, taskCh <-chan Task, resultCh chan<- Result) {
for i := 0; i < workers; i++ {
go func() {
for task := range taskCh {
select {
case <-task.Ctx.Done():
resultCh <- Result{TaskID: task.ID, Err: task.Ctx.Err()}
continue
default:
}</p><pre class="brush:php;toolbar:false;">            // 模拟处理
            output := strings.ToUpper(string(task.Data))
            resultCh <- Result{TaskID: task.ID, Output: output}
        }
    }()
}

}