前言

在日常开发中, 经常会使用chan来进行协程之间的通信. 对chan的操作也无外乎读写关. 而本次, 就是从chan的关闭而来.

假设我们对外提供的方法如下:

type Chan struct {
	ch chan int
}

func (c *Chan) Close() {
	close(c.ch)
}

func (c *Chan) Send(v int) {
	c.ch <- v
}

那么, 简单写段逻辑测试一下:

func main() {
	for {
		c := &Chan{ch: make(chan int, 100)}
		w := sync.WaitGroup{}
		w.Add(2)
		go func() {
			defer w.Done()
			c.Send(1)
		}()
		go func() {
			defer w.Done()
			c.Close()
		}()
		w.Wait()
	}
}

很快, 你就会看到报错了:

报错的原因也很简单, 就是因为向已关闭的管道中写数据.

我们如何能实现安全的管道关闭操作呢?

注意, 本次不讨论重复调用Close方法的情况, 这种只要加锁保证单次执行就可以了, 情况简单.

如何关闭管道

1.漏洞百出版

有的小伙伴想到了, 加个状态判断不就行了, 管道关闭之后就不再往里写咯. 并很快给出了代码:

type Chan struct {
	ch       chan int
	isClosed bool
}
func (c *Chan) Close() {
	c.isClosed = true
	close(c.ch)
}
func (c *Chan) Send(v int) bool {
	if c.isClosed {
		return false
	}
	c.ch <- v
	return true
}

但是, 我相信在运行之后就不会这么想了.

问题: 如果在发送时的 isClosed判断与chan通信之间发生了管道的关闭, 还是会出错.

2. 粗糙的正确加锁版

既然发生错误的原因是操作的原则性, 那很自然的就会想到加锁.

type Chan struct {
	ch       chan int
	isClosed bool
	lock     sync.Mutex
}

func (c *Chan) Close() {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.isClosed = true
	close(c.ch)
}

func (c *Chan) Send(v int) bool {
	c.lock.Lock()
	defer c.lock.Unlock()
	if c.isClosed {
		return false
	}
	c.ch <- v
	return true
}

将isClosed变量的访问与赋值通过加锁来实现原子操作, 那么我们直接让Close 和Send函数全体加锁, 将其强行改为串行, 不就解决了嘛?

开心的告诉你, 没错, 这样确实解决了.而且, 加锁的粒度不能再小了,

但是, 不得不遗憾的告诉你, 这么写只能说看起来对.

问题: 因为向管道写数据是阻塞的, 当chan满了再写数据, 就会阻塞在这里, 持有的锁就不会释放. 导致关闭函数一直无法执行.

有的小机灵鬼想到了, 使用select-case将阻塞写改为非阻塞写不就行了嘛? 确实可以, 但这样无疑会增加调用者的成本.

3. 小机灵版本

有的小机灵想到了, 我再Send的时候把panic抓住处理掉不就行了么.

type Chan struct {
	ch chan int
}

func (c *Chan) Close() {
	close(c.ch)
}

func (c *Chan) Send(v int) (ret bool) {
	defer func() {
		if r := recover(); r != nil {
			ret = false
		}
	}()
	c.ch <- v
	return true
}

开心的告诉你, 确实可以. 而且不存在锁的竞争问题, 很好.

但是, 在Go的哲学中, 流程中的异常是通过error返回, 通过捕捉panic来实现总觉得怪怪的.

4. Sender关闭

既然总会遇到风险, 那么我在Send汉中中进行close可以么?

type Chan struct {
	ch        chan int
	needClose bool
	once      sync.Once
}

func (c *Chan) Close() {
	c.needClose = true
}

func (c *Chan) Send(v int) (ret bool) {
	if c.needClose {
		c.once.Do(func() {
			close(c.ch)
		})
		return false
	}
	c.ch <- v
	return true
}

这样确实也是可以的, 但是别高兴的太早.

问题: 如果在Close之后没有调用Send方法, 那么此chan就不会关闭, 也就无法通知到接收方了. 因此, 十分不推荐.

包括使用额外的chan来实现, 也不推荐, 因为没有将chan关闭, 无法通知接收方(如下):

type Chan struct {
	ch   chan int
	done chan int
}

func (c *Chan) Close() {
	close(c.done)
}

func (c *Chan) Send(v int) (ret bool) {
	select {
	case <-c.done:
		return false
	case c.ch <- v:
		return true
	}
}

等等吧, 其实有很多方式来实现, 在这里就不一一赘述了.

总结

回过头来再想.

1. 为什么在官方的设计中, 管道不能重复关闭?
2. 为什么向已关闭的管道发送数据会panic, 却可以从已关闭的管道读取数据?

从官方 API 的设计中, 我们可以猜到其希望调用方做的事情:

1. 明确的知道自己什么时候关闭管道, 且仅有一个人来关闭
2. 发送方明确的知道管道是否已经关闭, 所以在管道关闭后不会再写数据
3. 接收方不知道管道的关闭情况, 因此需要通过返回值来判断

而我们关闭管道的目的是什么呢? 无非是:

1. 通知接收方, 管道已经关闭, 处理完管道中的余量就可以退出了
2. 通知发送方, 不要再发送数据量

那么, 我们就目的而言来分析:

1. 通知接收方, 这个在官方的设计中就已经支持了, 只要保证管道不会重复close就行
2. 通知发送方. 这里我们分两种情况来看:
   1. 后续没有数据需要发送. 此时貌似也不需要通知哈.
   2. 还有数据没有发送出去. 若还有数据的话, 不发出去是否会有问题? 是否应该等数据发完再关闭? 后续没发出去的数据如何处理?

综上, 我得出如下结论, 而这想必也是官方的意思吧:

1. 管道的关闭应由发送方负责
2. 发送方需在确认所有数据都发出之后再关闭管道

那么, 如果真的真的就是要在发送完之前关闭管道怎么办呢?

• 自查是否是设计问题, 导致管道的关闭者无法获得数据的发送信息
• 若流程实在无法更改, 推荐使用一个第三者(管道或锁)来将关闭状态通知给发送方和接收方, 而不关闭使用中的管道
  • 因为管道本身就已经加锁了, 如果再通过加锁来实现的话无疑会造成额外的性能损耗. 甚至于我认为recover都要被加锁更好.
  • 当然了, 如果能通过修改设计来保证的话就更好了

以上, 如果你有什么其他意见, 烦请不吝赐教

---

原文地址: https://hujingnb.com/archives/888