客户端同步中的竞争条件

Question

我有一个 Web 应用程序，其服务器为每个 websocket 连接创建一个客户端。客户端充当 websocket 连接和 Hub 的单个实例之间的中介。 Hub 维护一组注册的客户端并向客户端广播消息。这工作得很好，但问题是客户端可能会错过从服务器生成客户端在连接时接收的初始状态包到客户端注册到集线器并开始接收广播事件之间的事件。

我的想法是在从数据库中获取任何信息之前向集线器注册客户端。这将确保客户端不会错过任何广播，尽管现在它可以接收已经应用于它接收的初始状态的消息。为了让客户端忽略这些消息，我可以在初始状态包和广播事件中包含一个单调的时间戳。

你能想出更优雅/更简单的解决方案吗？

Answer 1

我过去曾使用预写日志来执行此类操作。简而言之，在集线器中保留消息的环形缓冲区。然后重播在新客户端初始化时发送到现有客户端的消息。

如果您愿意，您也可以向客户公开这个概念。这样您就可以实现高效的重新连接（尤其适用于移动连接）。当客户端断开 websocket 连接时，他们可以重新连接并说“嘿，又是我。看起来我们被打断了。我看到的最后一条消息是 42 号。有什么新消息？”

以下内容来自记忆，因此仅作为想法的说明，而不是完成的实现。例如，为了简洁起见，我省略了 client.send 周围的选择语句。

package main

import (
    "container/list"
    "sync"

    "github.com/gorilla/websocket"
)

type Client struct { // all unchanged
    hub  *Hub
    conn *websocket.Conn
    send chan []byte
}

type Hub struct {
    mu      *sync.RWMutex
    wal     list.List        // List if recent messages
    clients map[*Client]bool // Registered clients.

    register chan Registration // not a chan *Client anymore

    broadcast  chan []byte
    unregister chan *Client
}

type Registration struct {
    client *Client

    // init is a function that is executed before the client starts to receive
    // broadcast messages. All messages that are broadcast while init is
    // running will be sent after init returns.
    init func()
}

func (h *Hub) run() {
    for {
        select {
        case reg := <-h.register:
            // Take note of the most recent message as of right now. 
            // initClient will replay all later messages
            h.mu.RLock()
            head := h.wal.Back()
            h.mu.RUnlock()

            go h.initClient(reg, head)
        case client := <-h.unregister:
            h.mu.Lock()
            if _, ok := h.clients[client]; ok {
                delete(h.clients, client)
                close(client.send)
            }
            h.mu.Unlock()
        case message := <-h.broadcast:
            h.mu.Lock()
            h.wal.PushBack(message)
            // TODO: Trim list if too long by some metric (e.g. number of
            // messages, age, total message size, etc.)

            clients := make([]*Client, len(h.clients))
            copy(clients, h.clients)
            h.mu.Unlock()

            for client := range clients {
                // TODO: deal with backpressure
                client.send <- message
            }
        }
    }
}

func (h *Hub) initClient(reg Registration, head *list.Element) {
    reg.init()

    // send messages in h.wal after head
    for {
        h.mu.RLock()
        head = head.Next()
        if head == nil {
            // caught up
            h.clients[reg.client] = true
            h.mu.RUnlock()
            return
        }
        h.mu.RUnlock()

        // TODO: deal with backpressure
        reg.client.send <- head.Value.([]byte)
    }
}