Go語(yǔ)言實(shí)現(xiàn)一個(gè)簡(jiǎn)單生產(chǎn)者消費(fèi)者模型
簡(jiǎn)介:介紹生產(chǎn)者消費(fèi)者模型,及go簡(jiǎn)單實(shí)現(xià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è)你要寄一件快遞,大致過程如下。.
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ì)于郵寄快遞的人也沒有影響。假設(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)者的方法沒有返回之前,生產(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ù)過快,消費(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ī)制,先做后做 // 沒有數(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ù),如果沒有只能等待,也就是阻塞,直到管道被關(guān)閉。所以宏觀是生產(chǎn)者消費(fèi)者同步執(zhí)行。
另外:這里是只而外開辟一個(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ī)制,先做后做 // 沒有數(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í)訪問同一個(gè)公共區(qū)域,同時(shí)進(jìn)行不同的操作。都可以劃分為生產(chǎn)者消費(fèi)者模型,比如訂單系統(tǒng)。
很多用戶的訂單下達(dá)之后,放入緩沖區(qū)或者隊(duì)列中,然后系統(tǒng)從緩沖區(qū)中去讀來(lái)真正處理。系統(tǒng)不必開辟多個(gè)線程來(lái)對(duì)應(yīng)處理多個(gè)訂單,減少系統(tǒng)并發(fā)的負(fù)擔(dān)。通過生產(chǎn)者消費(fèi)者模式,將訂單系統(tǒng)與倉(cāng)庫(kù)管理系統(tǒng)隔離開,且用戶可以隨時(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é)果如下:
到此這篇關(guān)于Go語(yǔ)言實(shí)現(xiàn)一個(gè)簡(jiǎn)單生產(chǎn)者消費(fèi)者模型的文章就介紹到這了,更多相關(guān)Go語(yǔ)言生產(chǎn)者消費(fèi)者 內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Go語(yǔ)言并發(fā)之Sync包的6個(gè)關(guān)鍵概念總結(jié)
這篇文章主要為大家詳細(xì)介紹了Go語(yǔ)言并發(fā)中Sync包的6個(gè)關(guān)鍵概念,文中的示例代碼講解詳細(xì),對(duì)我們深入學(xué)習(xí)Go語(yǔ)言有一定的幫助,需要的可以參考一下2023-05-05Go位集合相關(guān)操作bitset庫(kù)安裝使用
這篇文章主要為大家介紹了Go位集合相關(guān)操作bitset庫(kù)安裝使用,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2022-07-07Golang分布式鎖簡(jiǎn)單案例實(shí)現(xiàn)流程
分布式鎖是控制分布式系統(tǒng)之間同步訪問共享資源的一種方式。如果不同的系統(tǒng)或是同一個(gè)系統(tǒng)的不同主機(jī)之間共享了一個(gè)或一組資源,那么訪問這些資源時(shí),需要通過一些互斥手段來(lái)防止彼此之間的干擾以保證一致性,在這種情況下,就需要使用分布式鎖了2022-12-12golang?http請(qǐng)求未釋放造成的錯(cuò)誤問題
這篇文章主要介紹了golang?http請(qǐng)求未釋放造成的錯(cuò)誤問題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-01-01