Go語言實現生產者-消費者問題的多種方法
生產者-消費者問題是并發編程中的經典問題，涉及多個生產者生成數據，多個消費者消費數據，二者通過緩沖區（隊列）進行協調，保證數據的正確傳遞和同步。本文將從簡單到復雜，使用不同的 Go 語言并發原語實現生產者-消費者模型，并詳細介紹所用知識點。
目錄
- 方法一：使用無緩沖 Channel（同步通信）
- 方法二：使用帶緩沖 Channel（異步通信）
- 方法三：使用 sync.Mutex + 條件變量 sync.Cond 實現緩沖區
- 方法四：使用 Channel + select 實現多路復用和超時控制
方法一：使用無緩沖 Channel（同步通信）
知識點
- 無緩沖 Channel：發送和接收必須同時準備好，適合嚴格同步的場景。
- Goroutine：輕量級線程，使用
go關鍵字啟動。 - sync.WaitGroup：等待所有 goroutine 完成。
代碼示例
package mainimport ("fmt""sync""time"
)func producer(id int, ch chan<- int, wg *sync.WaitGroup) {defer wg.Done()for i := 0; i < 3; i++ {item := id*100 + ifmt.Printf("生產者 %d 生產了產品 %d\n", id, item)ch <- item // 發送數據，阻塞直到有消費者接收time.Sleep(100 * time.Millisecond) // 模擬生產時間}
}func consumer(id int, ch <-chan int, wg *sync.WaitGroup) {defer wg.Done()for item := range ch {fmt.Printf("消費者 %d 消費了產品 %d\n", id, item)time.Sleep(150 * time.Millisecond) // 模擬消費時間}
}func main() {ch := make(chan int) // 無緩沖 channelvar wg sync.WaitGroup// 啟動生產者for i := 1; i <= 2; i++ {wg.Add(1)go producer(i, ch, &wg)}// 啟動消費者for i := 1; i <= 2; i++ {wg.Add(1)go consumer(i, ch, &wg)}// 等待生產者完成wg.Wait()// 關閉 channel，通知消費者結束close(ch)// 由于消費者在 range 中消費，關閉后會退出// 這里主 goroutine 退出，程序結束
}
說明
- 生產者發送數據時會阻塞，直到消費者接收，保證同步。
- 適合生產和消費速度相近的場景。
- 關閉 channel 后，消費者會自動退出。
方法二：使用帶緩沖 Channel（異步通信）
知識點
- 帶緩沖 Channel：允許生產者先發送一定數量數據，消費者稍后接收，提升并發效率。
- 生產者和消費者速度不匹配時，緩沖區能暫存數據，減少阻塞。
代碼示例
package mainimport ("fmt""math/rand""sync""time"
)func producer(id int, ch chan<- int, wg *sync.WaitGroup) {defer wg.Done()for i := 0; i < 5; i++ {item := id*100 + ifmt.Printf("生產者 %d 生產了產品 %d\n", id, item)ch <- itemtime.Sleep(time.Duration(rand.Intn(300)) * time.Millisecond)}
}func consumer(id int, ch <-chan int, wg *sync.WaitGroup) {defer wg.Done()for item := range ch {fmt.Printf("消費者 %d 消費了產品 %d\n", id, item)time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond)}
}func main() {rand.Seed(time.Now().UnixNano())ch := make(chan int, 3) // 帶緩沖 channel，緩沖區大小為3var wgProducers sync.WaitGroupvar wgConsumers sync.WaitGroup// 啟動生產者for i := 1; i <= 3; i++ {wgProducers.Add(1)go producer(i, ch, &wgProducers)}// 啟動消費者for i := 1; i <= 2; i++ {wgConsumers.Add(1)go consumer(i, ch, &wgConsumers)}// 等待所有生產者完成wgProducers.Wait()// 關閉 channel，通知消費者沒有更多數據close(ch)// 等待所有消費者完成wgConsumers.Wait()fmt.Println("所有生產者和消費者已完成工作，程序結束")
}
說明
- 生產者可以先發送數據到緩沖區，不必等待消費者立即接收。
- 緩沖區大小影響生產者和消費者的阻塞情況。
- 關閉 channel 后，消費者會自動退出。
方法三：使用 sync.Mutex + sync.Cond 實現緩沖區（手動實現隊列）
知識點
- sync.Mutex：互斥鎖，保護共享資源。
- sync.Cond：條件變量，支持等待和通知機制。
- 手動實現緩沖區：用切片模擬隊列，生產者和消費者通過條件變量協調。
代碼示例
package mainimport ("fmt""sync""time"
)type Buffer struct {items []intsize intlock sync.MutexnotEmpty *sync.CondnotFull *sync.Cond
}func NewBuffer(size int) *Buffer {b := &Buffer{items: make([]int, 0, size),size: size,}b.notEmpty = sync.NewCond(&b.lock)b.notFull = sync.NewCond(&b.lock)return b
}func (b *Buffer) Put(item int) {b.lock.Lock()defer b.lock.Unlock()// 如果緩沖區滿，等待 notFull 信號for len(b.items) == b.size {b.notFull.Wait()}b.items = append(b.items, item)fmt.Printf("生產了產品 %d，緩沖區大小: %d\n", item, len(b.items))// 通知消費者緩沖區非空b.notEmpty.Signal()
}func (b *Buffer) Get() int {b.lock.Lock()defer b.lock.Unlock()// 如果緩沖區空，等待 notEmpty 信號for len(b.items) == 0 {b.notEmpty.Wait()}item := b.items[0]b.items = b.items[1:]fmt.Printf("消費了產品 %d，緩沖區大小: %d\n", item, len(b.items))// 通知生產者緩沖區非滿b.notFull.Signal()return item
}func producer(id int, b *Buffer, count int, wg *sync.WaitGroup) {defer wg.Done()for i := 0; i < count; i++ {item := id*100 + ib.Put(item)time.Sleep(100 * time.Millisecond)}
}func consumer(id int, b *Buffer, wg *sync.WaitGroup, done <-chan struct{}) {defer wg.Done()for {select {case <-done:returndefault:item := b.Get()time.Sleep(150 * time.Millisecond)fmt.Printf("消費者 %d 處理了產品 %d\n", id, item)}}
}func main() {bufferSize := 5b := NewBuffer(bufferSize)var wgProducers sync.WaitGroupvar wgConsumers sync.WaitGroupdone := make(chan struct{})// 啟動生產者numProducers := 2produceCount := 10for i := 1; i <= numProducers; i++ {wgProducers.Add(1)go producer(i, b, produceCount, &wgProducers)}// 啟動消費者numConsumers := 3for i := 1; i <= numConsumers; i++ {wgConsumers.Add(1)go consumer(i, b, &wgConsumers, done)}// 等待生產者完成wgProducers.Wait()// 生產結束，等待緩沖區清空for {b.lock.Lock()empty := len(b.items) == 0b.lock.Unlock()if empty {break}time.Sleep(100 * time.Millisecond)}// 通知消費者退出close(done)// 等待消費者退出wgConsumers.Wait()fmt.Println("所有生產者和消費者已完成工作，程序結束")
}
說明
- 手動實現緩沖區，生產者和消費者通過條件變量等待和通知。
- 適合需要自定義緩沖區行為的場景。
- 需要額外處理消費者退出邏輯。
方法四：使用 Channel + select 實現多路復用和超時控制
知識點
- select：Go 語言中用于監聽多個 channel 的操作，支持超時和默認分支。
- 超時控制：防止 goroutine 永久阻塞。
- 多路復用：同時監聽多個事件。
代碼示例
package mainimport ("fmt""math/rand""time"
)func producer(id int, ch chan<- int, done <-chan struct{}) {for i := 0; i < 10; i++ {item := id*100 + iselect {case ch <- item:fmt.Printf("生產者 %d 生產了產品 %d\n", id, item)case <-done:fmt.Printf("生產者 %d 收到退出信號\n", id)return}time.Sleep(time.Duration(rand.Intn(300)) * time.Millisecond)}
}func consumer(id int, ch <-chan int, done <-chan struct{}) {for {select {case item, ok := <-ch:if !ok {fmt.Printf("消費者 %d 發現通道關閉，退出\n", id)return}fmt.Printf("消費者 %d 消費了產品 %d\n", id, item)time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond)case <-done:fmt.Printf("消費者 %d 收到退出信號\n", id)returncase <-time.After(2 * time.Second):fmt.Printf("消費者 %d 超時退出\n", id)return}}
}func main() {rand.Seed(time.Now().UnixNano())ch := make(chan int, 5)done := make(chan struct{})// 啟動生產者for i := 1; i <= 3; i++ {go producer(i, ch, done)}// 啟動消費者for i := 1; i <= 2; i++ {go consumer(i, ch, done)}// 運行一段時間后關閉生產者time.Sleep(5 * time.Second)close(done) // 通知所有 goroutine 退出// 關閉 channel，通知消費者沒有更多數據close(ch)// 主 goroutine 等待一段時間讓所有 goroutine 退出time.Sleep(3 * time.Second)fmt.Println("程序結束")
}說明
- 使用
select監聽多個 channel，支持超時和退出信號。 - 生產者和消費者都能響應退出通知，優雅結束。
- 適合復雜場景下的生產者-消費者模型。
總結
| 方法 | 復雜度 | 關鍵知識點 | 適用場景 |
|---|---|---|---|
| 方法一 | 簡單 | 無緩沖 channel，阻塞同步 | 生產消費速度相近，簡單同步 |
| 方法二 | 中等 | 帶緩沖 channel，異步通信 | 生產消費速度不匹配，提升效率 |
| 方法三 | 較復雜 | sync.Mutex + sync.Cond，手動緩沖區 | 需要自定義緩沖區行為，復雜同步 |
| 方法四 | 復雜 | select 多路復用，超時控制，退出通知 | 復雜場景，需多事件監聽和優雅退出 |
Go 語言提供了豐富的并發原語，能夠靈活實現生產者-消費者模型。根據實際需求和復雜度選擇合適的方法，能讓程序更高效、健壯。
![polarctf-web-[rce1]](http://pic.xiahunao.cn/polarctf-web-[rce1])
