登录
首页 >  Golang >  Go教程

再次探讨go实现无限 buffer 的 channel方法

来源:脚本之家

时间:2023-01-08 13:37:34 275浏览 收藏

亲爱的编程学习爱好者,如果你点开了这篇文章,说明你对《再次探讨go实现无限 buffer 的 channel方法》很感兴趣。本篇文章就来给大家详细解析一下,主要介绍一下channel、gobuffer,希望所有认真读完的童鞋们,都有实质性的提高。

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() 来拓展 sliceslice 的容量可以无限拓展下去(内存足够的话),所以 channel 也是 infiniteInfiniteChannel 的第一个成员就这么敲定下来的。

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 

其中, inChanoutChan 都是 unbuffered channel。

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

func (c *InfiniteChannel) background() {
	for true {
		select {
		case newVal := 

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() 的实现也非常简单。

// 取出队列的首个元素,如果队列为空,将会返回一个 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 


// out
012345678910111213141516171819
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 := 

outChan 进行一个简单封装

func (c *InfiniteChannel) background() {
	for true {
		select {
		case newVal := 

目前为止,一切照旧。

点睛之笔来了:

func (c *InfiniteChannel) outChanWrapper() chan interface{} {
	if len(c.data) == 0 {
		return nil
	}
	return c.outChan
}

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

background() 中,当执行到 case c.outChan 时,实际上将会变成:

case nil 

go 中,是无法往一个 nilchannel 中发送元素的。例如

func main() {
	var c chan interface{}
	select {
	case c 

因此,对于

select {
case newVal := 

将会一直阻塞在 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 := 
// 
// 
// 
func main() {
	c := make(chan interface{})
	close(c)
	for true {
		v, isOpen :=  false
//  false
//  false
//  false

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

func (c *InfiniteChannel) Close() {
	close(c.inChan)
}

func (c *InfiniteChannel) background() {
	for true {
		select {
		case newVal, isOpen := 

再测试一下

func main() {
	c := NewInfiniteChannel()
	go func() {
		for i := 0; i 


// output
012345678910111213141516171819
Process finished with the exit code 0

符合预期

遗憾

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

v,isOpen := 

可以通过 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 

参考

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

理论要掌握,实操不能落!以上关于《再次探讨go实现无限 buffer 的 channel方法》的详细介绍,大家都掌握了吧!如果想要继续提升自己的能力,那么就来关注golang学习网公众号吧!

声明:本文转载于:脚本之家 如有侵犯,请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>
评论列表