登录
首页 >  Golang >  Go问答

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

来源:stackoverflow

时间:2024-04-12 15:39:35 247浏览 收藏

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 退出。 updateclientselect 内部有一个退出通道,但 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,因为 listenupdateclient 都使用它,并且无法知道它们何时被另一个关闭。


解决方案


关闭连接以使 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学习网公众号下多多回复交流;文中若有不正之处,也希望回复留言以告知!

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