在我的代码中安全完成goroutine的正确方法是什么？

Question

我正在编写一个简单的tcp服务器，goroutine模型非常简单：

一个goroutine负责接受新的连接。对于每个新连接，将启动三个goroutine：

1. 一读
2. 一种用于处理和处理应用程序逻辑
3. 一个写

目前，一台服务器最多可服务1000个用户，因此我不会尝试限制goroutine的数量。

for {
    conn, err := listener.Accept()
    // ....
    connHandler := connHandler{
        conn:      conn,
        done:      make(chan struct{}),
        readChan:  make(chan string, 100),
        writeChan: make(chan string, 100),
    }
    // ....
    go connHandler.readAll()    
    go connHandler.processAll() 
    go connHandler.writeAll()   
}

我使用done通道通知所有三个通道完成，当用户注销或永久性网络错误发生时，done通道将被关闭（使用sync.Once确保关闭仅发生一次）：

func (connHandler *connHandler) Close() {
    connHandler.doOnce.Do(func() {
        connHandler.isClosed = true
        close(connHandler.done)
    })
}

下面是writeAll()方法的代码：

func (connHandler *connHandler) writeAll() {
    writer := bufio.NewWriter(connHandler.conn)

    for {
        select {
        case <-connHandler.done:
            connHandler.conn.Close()
            return
        case msg := <-connHandler.writeChan:
            connHandler.writeOne(msg, writer)
        }
    }
}

有一种Send通过向写通道发送字符串来向用户发送消息的方法：

func (connHandler *connHandler) Send(msg string) {
    case connHandler.writeChan <- msg:
}

该Send方法将主要在processAll()goroutine中被调用，但在许多其他goroutine中也将被调用，因为不同的用户需要彼此通信。

现在的问题是：如果用户A注销或网络失败，用户B向用户A发送一条消息，则用户B的goroutine可能会被永久阻止，因为没有人会收到来自该通道的消息。

我的解决方案：

我的第一个想法是使用一个布尔值来确保在发送给connHanler时未将其关闭：

func (connHandler *connHandler) Send(msg string) {
    if !connHandler.isClosed {
        connHandler.writeChan <- msg
    }
}

但是我认为connHandler.writeChan <- msg并且close(done)仍然可能同时发生，阻塞的可能性仍然存在。所以我必须添加一个超时：

func (connHandler *connHandler) Send(msg string) {
    if !connHandler.isClosed {
        timer := time.NewTimer(10 * time.Second)
        defer timer.Stop()
        select {
        case connHandler.writeChan <- msg:
        case <-timer.C:
            log.Warning(connHandler.Addr() + " send msg timeout:" + msg)
        }
    }

}

现在，我觉得代码很安全，但也很丑陋，每次发送消息时启动计时器都会带来不必要的开销。

然后，我阅读了这篇文章：https : //go101.org/article/channel-closing.html，我的问题看起来像文章中的第二个例子：

一个接收器，N个发送器，接收器通过关闭附加信号通道说“请停止发送更多”

但是我认为这种解决方案无法消除我遇到这种情况的可能性。

也许最简单的解决方案是只关闭写通道并使Send方法紧急，然后使用它recover来处理紧急情况？但是，这看起来也很丑陋。

那么，有没有简单直接的方法来完成我想做的事情？

（我的英语不好，所以如果有歧义，请指出，谢谢。）

Answer 1

你的例子看起来很不错，我认为你已经满足了 90% 的需要。

我认为您遇到的问题是发送，而您实际上可能已经“完成”了。

您可以使用“完成”通道来通知所有go 例程您已完成。您始终能够从关闭的通道读取值（它将是零值）。这意味着您可以更新您的Send(msg)方法以考虑已完成的通道。

func (connHandler *connHandler) Send(msg string) {
    select {
    case connHandler.writeChan <- msg:
    case <- connHandler.done:
        log.Debug("connHandler is done, exiting Send without sending.")
    case <-time.After(10 * time.Second):
        log.Warning(connHandler.Addr() + " send msg timeout:" + msg)
    }
}

现在此选择中将发生以下情况之一：

1. 消息发送至writeChan
2. close(done)已在其他地方调用，done chan 已关闭。您将能够从完成中读取内容，从而打破选择。
3. time.After(...) 将达到 10 秒，您将能够使发送超时。