当 websocket 连接关闭时关闭 redis 订阅并结束 go 例程

Question

我正在将事件从 redis 订阅推送到通过 websocket 连接的客户端。当客户端断开 websocket 时，我无法取消订阅和退出 redis go 例程。

受this post 的启发，这就是我目前所拥有的。我能够接收订阅事件并通过 websocket 向客户端发送消息，但是当客户端关闭 websocket 并且defer close(done) 代码触发时，我的case b, ok := <-done: 不会触发。默认情况下好像超载了？？？

package api

import (
    ...

    "github.com/garyburd/redigo/redis"
    "github.com/gorilla/websocket"
)

func wsHandler(w http.ResponseWriter, r *http.Request) {
    var upgrader = websocket.Upgrader{
        ReadBufferSize:  1024,
        WriteBufferSize: 1024,
    }

    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        HandleError(w, err)
        return
    }
    defer conn.Close()

    done := make(chan bool)
    defer close(done)

    for {
        var req WSRequest
        err := conn.ReadJSON(&req)
        if err != nil {
            HandleWSError(conn, err)
            return
        }
        defer conn.Close()

        go func(done chan bool, req *WSRequest, conn *websocket.Conn) {
            rc := redisPool.Get()
            defer rc.Close()

            psc := redis.PubSubConn{Conn: rc}
            if err := psc.PSubscribe(req.chanName); err != nil {
                HandleWSError(conn, err)
                return
            }
            defer psc.PUnsubscribe()

            for {
                select {
                case b, ok := <-done:
                    if !ok || b == true {
                        return
                    }
                default:
                    switch v := psc.Receive().(type) {
                    case redis.PMessage:
                        err := handler(conn, req, v)
                        if err != nil {
                            HandleWSError(conn, err)
                        }

                    case redis.Subscription:
                         log.Printf("%s: %s %d\n", v.Channel, v.Kind, v.Count)

                    case error:
                         log.Printf("error in redis subscription; err:\n%v\n", v)
                         HandleWSError(conn, v)

                    default:
                        // do nothing...
                        log.Printf("unknown redis subscription event type; %s\n", reflect.TypeOf(v))
                    }
                }
            }
        }(done, &req, conn)
    }
}

Answer 1

在为 websocket 连接提供服务后，进行这些更改以跳出读取循环：

• 维护为此 websocket 连接创建的 Redis 连接的一部分。

• 完成后取消订阅所有连接。

• 修改读取循环以在订阅计数为零时返回。

代码如下：

func wsHandler(w http.ResponseWriter, r *http.Request) {
    var upgrader = websocket.Upgrader{
        ReadBufferSize:  1024,
        WriteBufferSize: 1024,
    }

    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        HandleError(w, err)
        return
    }
    defer conn.Close()

    // Keep slice of all connections. Unsubscribe all connections on exit.
    var pscs []redis.PubSubConn
    defer func() {
        for _, psc := range rcs {
           psc.Unsubscribe() // unsubscribe with no args unsubs all channels
        }
    }()

    for {
        var req WSRequest
        err := conn.ReadJSON(&req)
        if err != nil {
            HandleWSError(conn, err)
            return
        }

        rc := redisPool.Get()
        psc := redis.PubSubConn{Conn: rc}
        pscs = append(pscs, psc)

        if err := psc.PSubscribe(req.chanName); err != nil {
            HandleWSError(conn, err)
            return
        }

        go func(req *WSRequest, conn *websocket.Conn) {
            defer rc.Close()
            for {
                switch v := psc.Receive().(type) {
                case redis.PMessage:
                    err := handler(conn, req, v)
                    if err != nil {
                        HandleWSError(conn, err)
                    }

                case redis.Subscription:
                     log.Printf("%s: %s %d\n", v.Channel, v.Kind, v.Count)
                     if v.Count == 0 {
                         return
                     }

                case error:
                     log.Printf("error in redis subscription; err:\n%v\n", v)
                     HandleWSError(conn, v)

                default:
                    // do nothing...
                    log.Printf("unknown redis subscription event type; %s\n", reflect.TypeOf(v))
                }
            }
        }(&req, conn)
    }
}

问题中的代码和这个答案为每个 websocket 客户端拨打多个 Redis 连接。一种更典型且可扩展的方法是跨多个客户端共享单个 Redis pubsub 连接。考虑到高级描述，典型方法可能适合您的应用程序，但鉴于问题中的代码，我仍然不确定您要做什么。