[[441586]]

本文介紹了生產(chǎn)者消費(fèi)者模型，和 go 實(shí)現(xiàn)的簡(jiǎn)單 demo。

一、生產(chǎn)者消費(fèi)者模型

生產(chǎn)者消費(fèi)者模型:某個(gè)模塊（函數(shù)等〉負(fù)責(zé)產(chǎn)生數(shù)據(jù)，這些數(shù)據(jù)由另一個(gè)模塊來(lái)負(fù)責(zé)處理（此處的模塊是廣義的，可以是類、函數(shù)、協(xié)程、線程、進(jìn)程等)。產(chǎn)生數(shù)據(jù)的模塊，就形象地稱為生產(chǎn)者;而處理數(shù)據(jù)的模塊，就稱為消費(fèi)者。

單單抽象出生產(chǎn)者和消費(fèi)者，還夠不上是生產(chǎn)者消費(fèi)者模型。該模式還需要有一個(gè)緩沖區(qū)處于生產(chǎn)者和消費(fèi)者之間，作為一個(gè)中介。生產(chǎn)者把數(shù)據(jù)放入緩沖區(qū)，而消費(fèi)者從緩沖區(qū)取出數(shù)據(jù)。大概的結(jié)構(gòu)如下圖。圖片

假設(shè)你要寄一件快遞，大致過(guò)程如下：

1．把快遞封好——相當(dāng)于生產(chǎn)者制造數(shù)據(jù)。

2．把快遞交給快遞中心——相當(dāng)于生產(chǎn)者把數(shù)據(jù)放入緩沖區(qū)。

3．郵遞員把快遞從快遞中心取出——相當(dāng)于消費(fèi)者把數(shù)據(jù)取出緩沖區(qū)。

這么看，有了緩沖區(qū)就有了以下好處：

解耦：降低消費(fèi)者和生產(chǎn)者之間的耦合度。有了快遞中心，就不必直接把快遞交給郵寄員，郵寄快遞的人不對(duì)郵寄員產(chǎn)生任何依賴，如果某一個(gè)天郵寄員換人了，對(duì)于郵寄快遞的人也沒(méi)有影響。假設(shè)生產(chǎn)者和消費(fèi)者分別是兩個(gè)類。如果讓生產(chǎn)者直接調(diào)用消費(fèi)者的某個(gè)方法，那么生產(chǎn)者對(duì)于消費(fèi)者就會(huì)產(chǎn)生依賴（也就是耦合)。將來(lái)如果消費(fèi)者的代碼發(fā)生變化，可能會(huì)真接影響到生產(chǎn)者。而如果兩者都依賴于某個(gè)緩沖區(qū)，兩者之間不直接依賴，耦合度也就相應(yīng)降低了。

并發(fā)：生產(chǎn)者消費(fèi)者數(shù)量不對(duì)等，依然能夠保持正常通信。由于函數(shù)調(diào)用是同步的(或者叫阻塞的)，在消費(fèi)者的方法沒(méi)有返回之前，生產(chǎn)者只好一直等在那邊。萬(wàn)一消費(fèi)者處理數(shù)據(jù)很慢，生產(chǎn)者只能等著浪費(fèi)時(shí)間。使用了生產(chǎn)者消費(fèi)者模式之后，生產(chǎn)者和消費(fèi)者可以是兩個(gè)獨(dú)立的并發(fā)主體。生產(chǎn)者把制造出來(lái)的數(shù)據(jù)往緩沖區(qū)一丟，就可以再去生產(chǎn)下一個(gè)數(shù)據(jù)?；旧喜挥靡蕾囅M(fèi)者的處理速度。郵寄快遞的人直接把快遞扔個(gè)快遞中心之后就不用管了。

緩存：生產(chǎn)者消費(fèi)者速度不匹配，暫存數(shù)據(jù)。如果郵寄快遞的人一次要郵寄多個(gè)快遞，那么郵寄員無(wú)法郵寄，就可以把其他的快遞暫存在快遞中心。也就是生產(chǎn)者短時(shí)間內(nèi)生產(chǎn)數(shù)據(jù)過(guò)快，消費(fèi)者來(lái)不及消費(fèi)，未處理的數(shù)據(jù)可以暫時(shí)存在緩沖區(qū)中。

二、Go語(yǔ)言實(shí)現(xiàn)

單向channel最典型的應(yīng)用是“生產(chǎn)者消費(fèi)者模型”。channel又分為有緩沖和無(wú)緩沖channel。channel中參數(shù)傳遞的時(shí)候，是作為引用傳遞。

1、無(wú)緩沖channel

示例代碼一實(shí)現(xiàn)如下 

package main  
import "fmt"  
func producer(out chan <- int) {  
 for i:=0; i<10; i++{  
  data := i*i 
  fmt.Println("生產(chǎn)者生產(chǎn)數(shù)據(jù):", data)  
  out <- data  // 緩沖區(qū)寫入數(shù)據(jù)  
 }  
 close(out)  //寫完關(guān)閉管道  
}   
func consumer(in <- chan int){  
        // 同樣讀取管道  
 //for{  
 // val, ok := <- in  
 // if ok {  
 //  fmt.Println("消費(fèi)者拿到數(shù)據(jù)：", data)  
 // }else{  
 //  fmt.Println("無(wú)數(shù)據(jù)")  
 //  break  
 // }  
 //}    
 // 無(wú)需同步機(jī)制，先做后做  
 // 沒(méi)有數(shù)據(jù)就阻塞等  
 for data := range in {  
  fmt.Println("消費(fèi)者得到數(shù)據(jù)：", data)  
 }  
}  
func main(){  
 // 傳參的時(shí)候顯式類型像隱式類型轉(zhuǎn)換，雙向管道向單向管道轉(zhuǎn)換  
 ch := make(chan int)  //無(wú)緩沖channel  
 go producer(ch)  // 子go程作為生產(chǎn)者  
 consumer(ch)  // 主go程作為消費(fèi)者 
 } 

這里使用無(wú)緩沖channel，生產(chǎn)者生產(chǎn)一次數(shù)據(jù)放入channel，然后消費(fèi)者從channel讀取數(shù)據(jù)，如果沒(méi)有只能等待，也就是阻塞，直到管道被關(guān)閉。所以宏觀是生產(chǎn)者消費(fèi)者同步執(zhí)行。

另外：這里是只而外開(kāi)辟一個(gè)go程執(zhí)行生產(chǎn)者，主go程執(zhí)行消費(fèi)者，如果也是用一個(gè)新的go程執(zhí)行消費(fèi)者，就需要阻塞main函數(shù)中的go程，否則不等待消費(fèi)者和生產(chǎn)者執(zhí)行完畢，主go程退出，程序直接結(jié)束，如示例代碼三。

生產(chǎn)者每一次生產(chǎn)，消費(fèi)者也只能拿到一次數(shù)據(jù)，緩沖區(qū)作用不大。結(jié)果如下：

2、有緩沖channel

示例代碼二如下 

package main  
import "fmt"  
func producer(out chan <- int) {  
 for i:=0; i<10; i++{  
  data := i*i  
  fmt.Println("生產(chǎn)者生產(chǎn)數(shù)據(jù):", data)  
  out <- data  // 緩沖區(qū)寫入數(shù)據(jù)  
 }  
 close(out)  //寫完關(guān)閉管道  
}   
func consumer(in <- chan int){  
 // 無(wú)需同步機(jī)制，先做后做  
 // 沒(méi)有數(shù)據(jù)就阻塞等  
 for data := range in {  
  fmt.Println("消費(fèi)者得到數(shù)據(jù)：", data)  
 }  
}  
func main(){  
 // 傳參的時(shí)候顯式類型像隱式類型轉(zhuǎn)換，雙向管道向單向管道轉(zhuǎn)換  
 ch := make(chan int, 5)  // 添加緩沖區(qū)，5  
 go producer(ch)  // 子go程作為生產(chǎn)者  
 consumer(ch)  // 主go程作為消費(fèi)者  
} 

有緩沖channel，只修改ch := make(chan int, 5) // 添加緩沖一句，只要緩沖區(qū)不滿，生產(chǎn)者可以持續(xù)向緩沖區(qū)channel放入數(shù)據(jù)，只要緩沖區(qū)不為空，消費(fèi)者可以持續(xù)從channel讀取數(shù)據(jù)。

就有了異步，并發(fā)的特性。結(jié)果如下：圖片

這里之所以終端生產(chǎn)者連續(xù)打印了大于緩沖區(qū)容量的數(shù)據(jù)，是因?yàn)榻K端打印屬于系統(tǒng)調(diào)用也是有延遲的，IO操作的時(shí)候，生產(chǎn)者同時(shí)向管道寫入，請(qǐng)求打印，管道的寫入讀取與終端輸出打印速度不匹配。

三、實(shí)際應(yīng)用

實(shí)際應(yīng)用中，同時(shí)訪問(wèn)同一個(gè)公共區(qū)域，同時(shí)進(jìn)行不同的操作。都可以劃分為生產(chǎn)者消費(fèi)者模型，比如訂單系統(tǒng)。很多用戶的訂單下達(dá)之后，放入緩沖區(qū)或者隊(duì)列中，然后系統(tǒng)從緩沖區(qū)中去讀來(lái)真正處理。

系統(tǒng)不必開(kāi)辟多個(gè)線程來(lái)對(duì)應(yīng)處理多個(gè)訂單，減少系統(tǒng)并發(fā)的負(fù)擔(dān)。通過(guò)生產(chǎn)者消費(fèi)者模式，將訂單系統(tǒng)與倉(cāng)庫(kù)管理系統(tǒng)隔離開(kāi)，且用戶可以隨時(shí)下單(生產(chǎn)數(shù)據(jù))。如果訂單系統(tǒng)直接調(diào)用倉(cāng)庫(kù)系統(tǒng)，那么用戶單擊下訂單按鈕后，要等到倉(cāng)庫(kù)系統(tǒng)的結(jié)果返回。這樣速度會(huì)很慢。

也就是：用戶變成了生產(chǎn)者，處理訂單管理系統(tǒng)變成了消費(fèi)者。

代碼示例三如下 

package main  
import (  
 "fmt"  
 "time"  
)  
// 模擬訂單對(duì)象  
type OrderInfo struct {  
 id int  
} 
// 生產(chǎn)訂單--生產(chǎn)者  
func producerOrder(out chan <- OrderInfo)  {  
 // 業(yè)務(wù)生成訂單  
 for i:=0; i<10; i++{  
  order := OrderInfo{id: i+1}  
  fmt.Println("生成訂單，訂單ID為：", order.id)  
  out <- order // 寫入channel  
 }  
 // 如果不關(guān)閉，消費(fèi)者就會(huì)一直阻塞，等待讀  
 close(out)  // 訂單生成完畢，關(guān)閉channel  
}  
// 處理訂單--消費(fèi)者  
func consumerOrder(in <- chan OrderInfo)  {  
 // 從channel讀取訂單，并處理  
 for order := range in{  
  fmt.Println("讀取訂單，訂單ID為：", order.id)  
 }  
}  
func main()  {  
 ch := make(chan OrderInfo, 5)  
 go producerOrder(ch)  
 go consumerOrder(ch)  
 time.Sleep(time.Second * 2)  
} 

這里如上面邏輯類似，不同的是用一個(gè)，OrderInfo結(jié)構(gòu)體模擬訂單作為業(yè)務(wù)處理對(duì)象。主線程使用time.Sleep(time.Second * 2)阻塞，否則，程序立即停止。結(jié)果如下：