总所周知，go 里面只有两种 channel，一种是 unbuffered channel, 其声明方式为

  1. ch := make(chan interface{})

另一种是 buffered channel，其声明方式为

  1. bufferSize := 5
  2. ch := make(chan interface{},bufferSize)

对于一个 buffered channel，无论它的 buffer 有多大，它终究是有极限的。这个极限就是该 channel 最初被 make 时，所指定的 bufferSize 。

  1. jojo,buffer channel 的大小是有极限的，我不做 channel 了。

一旦 channel 满了的话，再往里面添加元素的话，将会阻塞。

so how can we make a infinite buffer channel？

本文参考了 medinum 上面的一篇文章，有兴趣的同学可以直接阅读原文。

接口的设计

首先当然是建一个 struct，在百度翻译的帮助下，我们将这个 struct 取名为 InfiniteChannel

  1. type InfiniteChannel struct {
  2. }

思考一下 channel 的核心行为，实际上就两个，一个流入（Fan in）,一个流出（Fan out），因此我们添加如下几个 method。

  1. func (c *InfiniteChannel) In(val interface{}) {
  2. 	// todo
  3. }
  5. func (c *InfiniteChannel) Out() interface{} {
  6. 	// todo
  7. }
内部实现

通过 In() 接收的数据，总得需要一个地方来存放。我们可以用一个 slice 来存放，就算用 In() 往里面添加了很多元素，也可以通过 append() 来拓展 slice，slice 的容量可以无限拓展下去（内存足够的话），所以 channel 也是 infinite 。 InfiniteChannel 的第一个成员就这么敲定下来的。

  1. type InfiniteChannel struct {
  2. 	data    []interface{}
  3. }

用户调用 In() 和 Out() 时，可能是并发的环境，在 go 中如何进行并发编程，最容易想到的肯定是 channel 了，因此我们在内部准备两个 channel，一个 inChan，一个 outChan，用 inChan 来接收数据，用 outChan 来流出数据。

  1. type InfiniteChannel struct {
  2. 	inChan  chan interface{}
  3. 	outChan chan interface{}
  4. 	data    []interface{}
  5. }
  7. func (c *InfiniteChannel) In(val interface{}) {
  8. 	c.inChan <- val
  9. }
  11. func (c *InfiniteChannel) Out() interface{} {
  12. 	return <-c.outChan
  13. }

其中, InChan 和 outChan 都是 unbuffered channel。

此外，也肯定是需要一个 select 来处理来自 inChan 和 outChan 身上的事件。因此我们另起一个协程，在里面做 select 操作。

  1. func (c *InfiniteChannel) background() {
  2. 	for true {
  3. 		select {
  4. 		case newVal := <-c.inChan:
  5. 			c.data = append(c.data, newVal)
  6.         case c.outChan <- c.pop():		// pop() 将取出队列的首个元素
  7. 		}
  8. 	}
  9. }
  10. func NewInfiniteChannel() *InfiniteChannel {
  11. 	c := &InfiniteChannel{
  12. 		inChan:  make(chan interface{}),
  13. 		outChan: make(chan interface{}),
  14. 	}
  15. 	go c.background()	// 注意这里另起了一个协程
  16. 	return c
  17. }

ps:感觉这也算是 go 并发编程的一个套路了。即

  1. 在 new struct 的时候，顺手 go 一个 select 协程，select 协程内执行一个 for 循环，不停的 select，监听一个或者多个 channel 的事件。
  2. struct 对外提供的 method，只会操作 struct 内的 channel（在本例中就是 inChan 和 outChan），不会操作 struct 内的其他数据（在本例中，In() 和 Out() 都没有直接操作 data）。
  3. 触发 channel 的事件后，由 select 协程进行数据的更新（在本例中就是 data ）。因为只有 select 协程对除 channel 外的数据成员进行读写操作，且 go 保证了对于 channel 的并发读写是安全的，所以代码是并发安全的。
  4. 如果 struct 是 exported ，用户或许会越过 new ，直接手动 make 一个 struct，可以考虑将 struct 设置为 unexported，把它的首字母小写即可。

pop() 的实现也非常简单。

  1. // 取出队列的首个元素，如果队列为空，将会返回一个 nil
  2. func (c *InfiniteChannel) pop() interface{} {
  3. 	if len(c.data) == 0 {
  4. 		return nil
  5. 	}
  6. 	val := c.data[0]
  7. 	c.data = c.data[1:]
  8. 	return val
  9. }
测试一下

用一个协程每秒钟生产一条数据，另一个协程每半秒消费一条数据，并打印。

  1. func main() {
  2. 	c := NewInfiniteChannel()
  3. 	go func() {
  4. 		for i := 0; i < 20; i++ {
  5. 			c.In(i)
  6. 			time.Sleep(time.Second)
  7. 		}
  8. 	}()
  10. 	for i := 0; i < 50; i++ {
  11. 		val := c.Out()
  12. 		fmt.Print(val)
  13. 		time.Sleep(time.Millisecond * 500)
  14. 	}
  15. }
  1. // out
  2. <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>
  3. Process finished with the exit code 0

可以看到，将 InfiniteChannel 内没有数据可供消费时，调用 Out() 将会返回一个 nil，不过这也在我们的意料之中，原因是 pop() 在队列为空时，将会返回 nil。

目前 InfiniteChannel 的行为与标准的 channel 的行为是有出入的，go 中的 channel，在没有数据却仍要取数据时会被阻塞，如何实现这个效果？

优化

我认为此处是是整篇文章最有技巧的地方，我第一次看到时忍不住拍案叫绝。

首先把原来的 background() 摘出来

  1. func (c *InfiniteChannel) background() {
  2. 	for true {
  3. 		select {
  4. 		case newVal := <-c.inChan:
  5. 			c.data = append(c.data, newVal)
  6. 		case c.outChan <- c.pop():
  7. 		}
  8. 	}
  9. }

对 outChan 进行一个简单封装

  1. func (c *InfiniteChannel) background() {
  2. 	for true {
  3. 		select {
  4. 		case newVal := <-c.inChan:
  5. 			c.data = append(c.data, newVal)
  6. 		case c.outChanWrapper() <- c.pop():
  7. 		}
  8. 	}
  9. }
  10. func (c *InfiniteChannel) outChanWrapper() chan interface{} {
  11. 	return c.outChan
  12. }

目前为止，一切照旧。

点睛之笔来了：

  1. func (c *InfiniteChannel) outChanWrapper() chan interface{} {
  2. 	if len(c.data) == 0 {
  3. 		return nil
  4. 	}
  5. 	return c.outChan
  6. }

在 c.data 为空的时候，返回一个 nil

在 background() 中，当执行到 case c.outChan <- c.pop(): 时，实际上将会变成：

  1. case nil <- nil:

在 go 中，是无法往一个 nil 的 channel 中发送元素的。例如

  1. func main() {
  2. 	var c chan interface{}
  3. 	select {
  4. 	case c <- 1:
  5. 	}
  6. }
  7. // fatal error: all goroutines are asleep - deadlock!
  8. func main() {
  9. 	var c chan interface{}
  10. 	select {
  11. 	case c <- 1:
  12. 	default:
  13. 		fmt.Println("hello world")
  14. 	}
  15. }
  16. // hello world

因此，对于

  1. select {
  2. case newVal := <-c.inChan:
  3. 	c.data = append(c.data, newVal)
  4. case c.outChanWrapper() <- c.pop():
  5. }

将会一直阻塞在 select 那里，直到 inChan 来了数据。

再测试一下
  1. 012345678910111213141516171819fatal error: all goroutines are asleep - deadlock!

最后，程序 panic 了，因为死锁了。

补充

实际上 channel 除了 In() 和 Out() 外，还有一个行为，即 close()，如果 channel close 后，依旧从其中取元素的话，将会取出该类型的默认值。

  1. func main() {
  2. 	c := make(chan interface{})
  3. 	close(c)
  4. 	for true {
  5. 		v := <-c
  6. 		fmt.Println(v)
  7. 		time.Sleep(time.Second)
  8. 	}
  9. }
  10. // output
  11. // <nil>
  12. // <nil>
  13. // <nil>
  14. // <nil>
  15. func main() {
  16. 	c := make(chan interface{})
  17. 	close(c)
  18. 	for true {
  19. 		v, isOpen := <-c
  20. 		fmt.Println(v, isOpen)
  21. 		time.Sleep(time.Second)
  22. 	}
  23. }
  24. // output
  25. // <nil> false
  26. // <nil> false
  27. // <nil> false
  28. // <nil> false

我们也需要实现相同的效果。

  1. func (c *InfiniteChannel) Close() {
  2. 	close(c.inChan)
  3. }
  5. func (c *InfiniteChannel) background() {
  6. 	for true {
  7. 		select {
  8. 		case newVal, isOpen := <-c.inChan:
  9. 			if isOpen {
  10. 				c.data = append(c.data, newVal)
  11. 			} else {
  12. 				c.isOpen = false
  13. 			}
  14. 		case c.outChanWrapper() <- c.pop():
  15. 		}
  16. 	}
  17. }
  19. func NewInfiniteChannel() *InfiniteChannel {
  20. 	c := &InfiniteChannel{
  21. 		inChan:  make(chan interface{}),
  22. 		outChan: make(chan interface{}),
  23. 		isOpen:  true,
  24. 	}
  25. 	go c.background()
  26. 	return c
  27. }
  29. func (c *InfiniteChannel) outChanWrapper() chan interface{} {
  30.     // 这里添加了对 c.isOpen 的判断
  31. 	if c.isOpen && len(c.data) == 0 {
  32. 		return nil
  33. 	}
  34. 	return c.outChan
  35. }
再测试一下
  1. func main() {
  2. 	c := NewInfiniteChannel()
  3. 	go func() {
  4. 		for i := 0; i < 20; i++ {
  5. 			c.In(i)
  6. 			time.Sleep(time.Second)
  7. 		}
  8. 		c.Close()		// 这里调用了 Close
  9. 	}()
  11. 	for i := 0; i < 50; i++ {
  12. 		val := c.Out()
  13. 		fmt.Print(val)
  14. 		time.Sleep(time.Millisecond * 500)
  15. 	}
  16. }
  1. // output
  2. 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>
  3. Process finished with the exit code 0

符合预期

遗憾

目前看上去已经很完美了，但是和标准的 channel 相比，仍然有差距。因为标准的 channel 是有这种用法的

  1. v,isOpen := <- ch

可以通过 isOpen 变量来获取 channel 的开闭情况。

因此 InfiniteChannel 也应该提供一个类似的 method

  1. func (c *InfiniteChannel) OutAndIsOpen() (interface{}, bool) {
  2. 	// todo
  3. }

可惜的是，要想得知 InfiniteChannel 是否是 Open 的，就必定要访问 InfiniteChannel 内的 isOpen 成员。

  1. type InfiniteChannel struct {
  2. 	inChan  chan interface{}
  3. 	outChan chan interface{}
  4. 	data    []interface{}
  5. 	isOpen  bool
  6. }

而 isOpen 并非 channel 类型，根据之前的套路，这种非 channel 类型的成员只应该被 select 协程访问。一旦有多个协程访问，就会出现并发问题，除非加锁。

我不能接受！所以干脆不提供这个 method 了，嘿嘿。

完整代码
  1. func main() {
  2. 	c := NewInfiniteChannel()
  3. 	go func() {
  4. 		for i := 0; i < 20; i++ {
  5. 			c.In(i)
  6. 			time.Sleep(time.Second)
  7. 		}
  8. 		c.Close()
  9. 	}()
  11. 	for i := 0; i < 50; i++ {
  12. 		val := c.Out()
  13. 		fmt.Print(val)
  14. 		time.Sleep(time.Millisecond * 500)
  15. 	}
  16. }
  18. type InfiniteChannel struct {
  19. 	inChan  chan interface{}
  20. 	outChan chan interface{}
  21. 	data    []interface{}
  22. 	isOpen  bool
  23. }
  25. func (c *InfiniteChannel) In(val interface{}) {
  26. 	c.inChan <- val
  27. }
  29. func (c *InfiniteChannel) Out() interface{} {
  30. 	return <-c.outChan
  31. }
  33. func (c *InfiniteChannel) Close() {
  34. 	close(c.inChan)
  35. }
  37. func (c *InfiniteChannel) background() {
  38. 	for true {
  39. 		select {
  40. 		case newVal, isOpen := <-c.inChan:
  41. 			if isOpen {
  42. 				c.data = append(c.data, newVal)
  43. 			} else {
  44. 				c.isOpen = false
  45. 			}
  46. 		case c.outChanWrapper() <- c.pop():
  47. 		}
  48. 	}
  49. }
  51. func NewInfiniteChannel() *InfiniteChannel {
  52. 	c := &InfiniteChannel{
  53. 		inChan:  make(chan interface{}),
  54. 		outChan: make(chan interface{}),
  55. 		isOpen:  true,
  56. 	}
  57. 	go c.background()
  58. 	return c
  59. }
  61. // 取出队列的首个元素，如果队列为空，将会返回一个 nil
  62. func (c *InfiniteChannel) pop() interface{} {
  63. 	if len(c.data) == 0 {
  64. 		return nil
  65. 	}
  66. 	val := c.data[0]
  67. 	c.data = c.data[1:]
  68. 	return val
  69. }
  71. func (c *InfiniteChannel) outChanWrapper() chan interface{} {
  72. 	if c.isOpen && len(c.data) == 0 {
  73. 		return nil
  74. 	}
  75. 	return c.outChan
  76. }

https://medium.com/capital-one-tech/building-an-unbounded-channel-in-go-789e175cd2cd

版权声明：本文为XiaoXiaoShuai-原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://www.cnblogs.com/XiaoXiaoShuai-/p/14878525.html