如何使用 for 循环在多个 goroutine 之间进行通信，并在其中一个 goroutine 内进行阻塞函数调用

Golang小白一枚，正在不断学习积累知识，现将学习到的知识记录一下，也是将我的所得分享给大家！而今天这篇文章《如何使用 for 循环在多个 goroutine 之间进行通信，并在其中一个 goroutine 内进行阻塞函数调用》带大家来了解一下##content_title##，希望对大家的知识积累有所帮助，从而弥补自己的不足，助力实战开发！

我正在编写一个 go 应用程序，它接受 websocket 连接，然后启动：

1. listen goroutine，监听连接上的客户端消息，并根据接收到的消息通过通道向 updateclient 发送客户端响应。
2. updateclient goroutine，用于写入连接。
3. processexternaldata goroutine，从消息队列接收数据，通过通道将数据发送给 updateclient，以便 updateclient 可以使用数据更新客户端。

我正在使用 gorilla 库进行 websocket 连接，并且它的读取调用被阻止。另外，它的 write 和 read 方法都不支持并发调用，这是我有 updateclient goroutine 的主要原因，它是调用 write 方法的单个例程。

当我需要关闭连接时就会出现问题，至少在两种情况下会发生：

1. 客户端关闭了连接或读取时发生错误。
2. processexternaldata 已完成，没有更多数据可更新客户端，应关闭连接。

因此 updateclient 需要以某种方式通知 listen 退出，反之亦然 listen 需要以某种方式通知 updateclient 退出。 updateclient 在 select 内部有一个退出通道，但 listen 不能有 select，因为它已经有一个 for 循环，并且内部有阻塞读取调用。

所以我所做的是在连接类型上添加了 isjobfinished 字段，这是 for 循环工作的条件：

type WsConnection struct {
    connection    *websocket.Conn
    writeChan     chan messageWithCb
    quitChan      chan bool
    isJobFinished bool
    userID        string
}

func processExternalData() {
    // receive data from message queue
    // send it to client via writeChan
}

func (conn *WsConnection) listen() {
    defer func() {
        conn.connection.Close()
        conn.quitChan <- true
    }()

    // keep the loop for communication with client
    for !conn.isJobFinished {
        _, message, err := conn.connection.ReadMessage()
        if err != nil {
            log.Println("read:", err)
            break

        }
        // convert message to type messageWithCb
        switch clientMessage.MessageType {
        case userNotFound:
            conn.writeChan <- messageWithCb{
                message: map[string]interface{}{
                    "type":    user,
                    "payload": false,
                },
            }
        default:
            log.Printf("Unknown message type received: %v", clientMessage)
        }
    }
    log.Println("end of listen")
}

func updateClient(w http.ResponseWriter, req *http.Request) {
    upgrader.CheckOrigin = func(req *http.Request) bool {
        return true
    }
    connection, err := upgrader.Upgrade(w, req, nil)
    if err != nil {
        log.Print("upgrade:", err)
        return
    }
    wsConn := &WsConnection{
        connection: connection,
        writeChan:  make(chan messageWithCb),
        quitChan:   make(chan bool),
    }
    go wsConn.listen()
    for {
        select {
        case msg := <-wsConn.writeChan:
            err := connection.WriteJSON(msg.message)
            if err != nil {
                log.Println("connection.WriteJSON error: ", err)
            }
            if wsConn.isJobFinished {
                if msg.callback != nil {
                    msg.callback() // sends on `wsConn.quitChan` from a goroutine
                }
            }
        case <-wsConn.quitChan:
            // clean up
            wsConn.connection.Close()
            close(wsConn.writeChan)
            return
        }
    }
}

我想知道 go 中是否存在针对此类情况的更好模式。具体来说，我希望能够在 listen 内部有一个退出通道，以便 updateclient 可以通知它退出，而不是维护 isjobfinished 字段。此外，在这种情况下，不保护 isjobfinished 字段也没有危险，因为只有一种方法写入它，但如果逻辑变得更复杂，则必须保护 listen 中的 for 循环内的字段可能会对性能产生负面影响。

此外，我无法关闭 quitechan，因为 listen 和 updateclient 都使用它，并且无法知道它们何时被另一个关闭。

解决方案

关闭连接以使 listen goroutine 脱离阻塞读取调用。

在updateclient中，添加defer语句，用于关闭连接并清理其他资源。出现任何错误或来自退出通道的通知时从函数返回：

updateclient(w http.responsewriter, req *http.request) {
    upgrader.checkorigin = func(req *http.request) bool {
        return true
    }
    connection, err := upgrader.upgrade(w, req, nil)
    if err != nil {
        log.print("upgrade:", err)
        return
    }
    defer connection.close() // <--- add this line
    wsconn := &wsconnection{
        connection: connection,
        writechan:  make(chan messagewithcb),
        quitchan:   make(chan bool),
    }
    defer close(writechan) // <-- cleanup moved out of loop below.
    go wsconn.listen()
    for {
        select {
        case msg := <-wsconn.writechan:
            err := connection.writejson(msg.message)
            if err != nil {
                log.println("connection.writejson error: ", err)
                return
            }
        case <-wsconn.quitchan:
            return
        }
    }
}

在 listen 函数中，循环直到读取连接时出错。当 updateclient 关闭连接时，连接上的读取立即返回并出现错误。

为了防止 updateclient 首先返回的情况下 listen 永远阻塞，请关闭退出通道而不是发送值。

func (conn *WsConnection) listen() {
    defer func() {
        conn.connection.Close()
        close(conn.quitChan) // <-- close instead of sending value
    }()

    // keep the loop for communication with client
    for  {
        _, message, err := conn.connection.ReadMessage()
        if err != nil {
            log.Println("read:", err)
            break

        }
        // convert message to type messageWithCb
        switch clientMessage.MessageType {
        case userNotFound:
            conn.writeChan <- messageWithCb{
                message: map[string]interface{}{
                    "type":    user,
                    "payload": false,
                },
            }
        default:
            log.Printf("Unknown message type received: %v", clientMessage)
        }
    }
    log.Println("end of listen")
}

不需要字段 isjobfinished。

问题和此答案中的代码的一个问题是 writechan 的关闭与发送到通道不协调。如果没有看到 processexternaldata 函数，我无法评论此问题的解决方案。

使用互斥体而不是 goroutine 来限制写入并发可能是有意义的。同样，需要 processexternaldata 函数中的代码来进一步评论此主题。

今天关于《如何使用 for 循环在多个 goroutine 之间进行通信，并在其中一个 goroutine 内进行阻塞函数调用》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！