再次探討go實現無限 buffer 的 channel方法
前言
總所周知,go
裡面隻有兩種 channel
,一種是 unbuffered channel, 其聲明方式為
ch := make(chan interface{})
另一種是 buffered channel,其聲明方式為
bufferSize := 5 ch := make(chan interface{},bufferSize)
對於一個 buffered channel,無論它的 buffer 有多大,它終究是有極限的。這個極限就是該 channel 最初被 make 時,所指定的 bufferSize 。
jojo,buffer channel 的大小是有極限的,我不做 channel 瞭。
一旦 channel
滿瞭的話,再往裡面添加元素的話,將會阻塞。
so how can we make a infinite buffer channel?
本文參考瞭 medinum 上面的一篇文章,有興趣的同學可以直接閱讀原文。
實現
接口的設計
首先當然是建一個 struct
,在百度翻譯的幫助下,我們將這個 struct
取名為 InfiniteChannel
type InfiniteChannel struct { }
思考一下 channel
的核心行為,實際上就兩個,一個流入(Fan in),一個流出(Fan out),因此我們添加如下幾個 method。
func (c *InfiniteChannel) In(val interface{}) { // todo } func (c *InfiniteChannel) Out() interface{} { // todo }
內部實現
通過 In()
接收的數據,總得需要一個地方來存放。我們可以用一個 slice
來存放,就算用 In()
往裡面添加瞭很多元素,也可以通過 append()
來拓展 slice
,slice
的容量可以無限拓展下去(內存足夠的話),所以 channel
也是 infinite
。 InfiniteChannel
的第一個成員就這麼敲定下來的。
type InfiniteChannel struct { data []interface{} }
用戶調用 In()
和 Out()
時,可能是並發的環境,在 go
中如何進行並發編程,最容易想到的肯定是 channel
瞭,因此我們在內部準備兩個 channel
,一個 inChan
,一個 outChan
,用 inChan
來接收數據,用 outChan
來流出數據。
type InfiniteChannel struct { inChan chan interface{} outChan chan interface{} data []interface{} } func (c *InfiniteChannel) In(val interface{}) { c.inChan <- val } func (c *InfiniteChannel) Out() interface{} { return <-c.outChan }
其中, inChan
和 outChan
都是 unbuffered channel。
此外,也肯定是需要一個 select
來處理來自 inChan
和 outChan
身上的事件。因此我們另起一個協程,在裡面做 select
操作。
func (c *InfiniteChannel) background() { for true { select { case newVal := <-c.inChan: c.data = append(c.data, newVal) case c.outChan <- c.pop(): // pop() 將取出隊列的首個元素 } } } func NewInfiniteChannel() *InfiniteChannel { c := &InfiniteChannel{ inChan: make(chan interface{}), outChan: make(chan interface{}), } go c.background() // 註意這裡另起瞭一個協程 return c }
ps:感覺這也算是 go 並發編程的一個套路瞭。即
- 在 new struct 的時候,順手 go 一個 select 協程,select 協程內執行一個 for 循環,不停的 select,監聽一個或者多個 channel 的事件。
- struct 對外提供的 method,隻會操作 struct 內的 channel(在本例中就是 inChan 和 outChan),不會操作 struct 內的其他數據(在本例中,In() 和 Out() 都沒有直接操作 data)。
- 觸發 channel 的事件後,由 select 協程進行數據的更新(在本例中就是 data )。因為隻有 select 協程對除 channel 外的數據成員進行讀寫操作,且 go 保證瞭對於 channel 的並發讀寫是安全的,所以代碼是並發安全的。
- 如果 struct 是 exported ,用戶或許會越過 new ,直接手動 make 一個 struct,可以考慮將 struct 設置為 unexported,把它的首字母小寫即可。
pop()
的實現也非常簡單。
// 取出隊列的首個元素,如果隊列為空,將會返回一個 nil func (c *InfiniteChannel) pop() interface{} { if len(c.data) == 0 { return nil } val := c.data[0] c.data = c.data[1:] return val }
測試一下
用一個協程每秒鐘生產一條數據,另一個協程每半秒消費一條數據,並打印。
func main() { c := NewInfiniteChannel() go func() { for i := 0; i < 20; i++ { c.In(i) time.Sleep(time.Second) } }() for i := 0; i < 50; i++ { val := c.Out() fmt.Print(val) time.Sleep(time.Millisecond * 500) } }
// out <nil>0<nil>1<nil>23<nil>4<nil><nil>5<nil>67<nil><nil>89<nil><nil>1011<nil>12<nil>13<nil>14<nil>15<nil>16<nil>17<nil><nil>1819<nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil> Process finished with the exit code 0
可以看到,將 InfiniteChannel
內沒有數據可供消費時,調用 Out()
將會返回一個 nil
,不過這也在我們的意料之中,原因是 pop()
在隊列為空時,將會返回 nil。
目前 InfiniteChannel
的行為與標準的 channel
的行為是有出入的,go
中的 channel
,在沒有數據卻仍要取數據時會被阻塞,如何實現這個效果?
優化
我認為此處是是整篇文章最有技巧的地方,我第一次看到時忍不住拍案叫絕。
首先把原來的 background()
摘出來
func (c *InfiniteChannel) background() { for true { select { case newVal := <-c.inChan: c.data = append(c.data, newVal) case c.outChan <- c.pop(): } } }
對 outChan
進行一個簡單封裝
func (c *InfiniteChannel) background() { for true { select { case newVal := <-c.inChan: c.data = append(c.data, newVal) case c.outChanWrapper() <- c.pop(): } } } func (c *InfiniteChannel) outChanWrapper() chan interface{} { return c.outChan }
目前為止,一切照舊。
點睛之筆來瞭:
func (c *InfiniteChannel) outChanWrapper() chan interface{} { if len(c.data) == 0 { return nil } return c.outChan }
在 c.data
為空的時候,返回一個 nil
在 background()
中,當執行到 case c.outChan <- c.pop():
時,實際上將會變成:
case nil <- nil:
在 go
中,是無法往一個 nil
的 channel
中發送元素的。例如
func main() { var c chan interface{} select { case c <- 1: } } // fatal error: all goroutines are asleep - deadlock! func main() { var c chan interface{} select { case c <- 1: default: fmt.Println("hello world") } } // hello world
因此,對於
select { case newVal := <-c.inChan: c.data = append(c.data, newVal) case c.outChanWrapper() <- c.pop(): }
將會一直阻塞在 select
那裡,直到 inChan
來瞭數據。
再測試一下
012345678910111213141516171819fatal error: all goroutines are asleep - deadlock!
最後,程序 panic
瞭,因為死鎖瞭。
補充
實際上 channel
除瞭 In()
和 Out()
外,還有一個行為,即 close()
,如果 channel close 後,依舊從其中取元素的話,將會取出該類型的默認值。
func main() { c := make(chan interface{}) close(c) for true { v := <-c fmt.Println(v) time.Sleep(time.Second) } } // output // <nil> // <nil> // <nil> // <nil> func main() { c := make(chan interface{}) close(c) for true { v, isOpen := <-c fmt.Println(v, isOpen) time.Sleep(time.Second) } } // output // <nil> false // <nil> false // <nil> false // <nil> false
我們也需要實現相同的效果。
func (c *InfiniteChannel) Close() { close(c.inChan) } func (c *InfiniteChannel) background() { for true { select { case newVal, isOpen := <-c.inChan: if isOpen { c.data = append(c.data, newVal) } else { c.isOpen = false } case c.outChanWrapper() <- c.pop(): } } } func NewInfiniteChannel() *InfiniteChannel { c := &InfiniteChannel{ inChan: make(chan interface{}), outChan: make(chan interface{}), isOpen: true, } go c.background() return c } func (c *InfiniteChannel) outChanWrapper() chan interface{} { // 這裡添加瞭對 c.isOpen 的判斷 if c.isOpen && len(c.data) == 0 { return nil } return c.outChan }
再測試一下
func main() { c := NewInfiniteChannel() go func() { for i := 0; i < 20; i++ { c.In(i) time.Sleep(time.Second) } c.Close() // 這裡調用瞭 Close }() for i := 0; i < 50; i++ { val := c.Out() fmt.Print(val) time.Sleep(time.Millisecond * 500) } }
// output 012345678910111213141516171819<nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil><nil> Process finished with the exit code 0
符合預期
遺憾
目前看上去已經很完美瞭,但是和標準的 channel
相比,仍然有差距。因為標準的 channel
是有這種用法的
v,isOpen := <- ch
可以通過 isOpen
變量來獲取 channel
的開閉情況。
因此 InfiniteChannel
也應該提供一個類似的 method
func (c *InfiniteChannel) OutAndIsOpen() (interface{}, bool) { // todo }
可惜的是,要想得知 InfiniteChannel
是否是 Open
的,就必定要訪問 InfiniteChannel
內的 isOpen
成員。
type InfiniteChannel struct { inChan chan interface{} outChan chan interface{} data []interface{} isOpen bool }
而 isOpen
並非 channel
類型,根據之前的套路,這種非 channel
類型的成員隻應該被 select
協程訪問。一旦有多個協程訪問,就會出現並發問題,除非加鎖。
我不能接受!所以幹脆不提供這個 method 瞭,嘿嘿。
完整代碼
func main() { c := NewInfiniteChannel() go func() { for i := 0; i < 20; i++ { c.In(i) time.Sleep(time.Second) } c.Close() }() for i := 0; i < 50; i++ { val := c.Out() fmt.Print(val) time.Sleep(time.Millisecond * 500) } } type InfiniteChannel struct { inChan chan interface{} outChan chan interface{} data []interface{} isOpen bool } func (c *InfiniteChannel) In(val interface{}) { c.inChan <- val } func (c *InfiniteChannel) Out() interface{} { return <-c.outChan } func (c *InfiniteChannel) Close() { close(c.inChan) } func (c *InfiniteChannel) background() { for true { select { case newVal, isOpen := <-c.inChan: if isOpen { c.data = append(c.data, newVal) } else { c.isOpen = false } case c.outChanWrapper() <- c.pop(): } } } func NewInfiniteChannel() *InfiniteChannel { c := &InfiniteChannel{ inChan: make(chan interface{}), outChan: make(chan interface{}), isOpen: true, } go c.background() return c } // 取出隊列的首個元素,如果隊列為空,將會返回一個 nil func (c *InfiniteChannel) pop() interface{} { if len(c.data) == 0 { return nil } val := c.data[0] c.data = c.data[1:] return val } func (c *InfiniteChannel) outChanWrapper() chan interface{} { if c.isOpen && len(c.data) == 0 { return nil } return c.outChan }
參考
https://medium.com/capital-one-tech/building-an-unbounded-channel-in-go-789e175cd2cd
以上就是再次探討go實現無限 buffer 的 channel方法的詳細內容,更多關於go無限 buffer 的 channel的資料請關註WalkonNet其它相關文章!
推薦閱讀:
- Go select使用與底層原理講解
- Go緩沖channel和非緩沖channel的區別說明
- go select的用法
- Golang的select多路復用及channel使用操作
- go語言中如何使用select的實現示例