node.js - Websocket 传输可靠性(重新连接期间的 Socket.io 数据丢失)

Question

二手

NodeJS、Socket.io

问题

假设有 2 个用户 U1 和 U2，通过 Socket.io 连接到一个应用程序。算法如下:

1. U1 完全断开互联网连接(例如关闭互联网)
2. U2 向 U1 发送消息。
3. U1 尚未收到消息，因为 Internet 已关闭
4. 服务器检测U1因心跳超时而断开连接
5. U1 重新连接到 socket.io
6. U1 从未收到来自 U2 的消息 - 我猜它在第 4 步丢失了。

可能的解释

我想我明白为什么会这样:

• 在第 4 步中，Server 也会杀死套接字实例和发送到 U1 的消息队列
• 此外，在第 5 步 U1 和 Server 会创建新连接(它不会被重复使用)，因此即使消息仍在排队，之前的连接也会丢失。

需要帮助

如何防止此类数据丢失？我必须使用心跳，因为我不会让人们永远卡在应用程序中。此外，我仍然必须提供重新连接的可能性，因为当我部署新版本的应用程序时，我希望零停机时间。

附:我称之为“消息”的东西不仅仅是我可以存储在数据库中的文本消息，而是有值(value)的系统消息，必须保证传递，否则 UI 会搞砸。

谢谢!

---

加法1

我已经有一个用户帐户系统。而且，我的应用程序已经很复杂了。添加离线/在线状态无济于事，因为我已经有了这种东西。问题是不同的。

查看第 2 步。在此步骤中，从技术上讲，我们不能说 U1 是否离线，他只是失去连接，可以说 2 秒钟，可能是因为网络不好。所以 U2 向他发送了一条消息，但 U1 没有收到它，因为他的互联网仍然关闭(第 3 步)。需要第 4 步来检测离线用户，假设超时为 60 秒。最终，再过 10 秒，U1 的互联网连接就建立起来了，他重新连接到了 socket.io。但是来自 U2 的消息在空间中丢失了，因为服务器 U1 因超时而断开连接。

这就是问题所在，我不想 100% 交付。

---

解决方案

1. 在 {} 用户中收集发射(发射名称和数据)，由随机发射 ID 标识。发送发射
2. 在客户端确认发射(使用 emitID 将发射发送回服务器)
3. 如果确认 - 从由 emitID 标识的 {} 中删除对象
4. 如果用户重新连接 - 检查此用户的 {} 并循环遍历它，为 {} 中的每个对象执行第 1 步
5. 当断开连接或/和连接时，如有必要，为用户刷新 {}

// Server
const pendingEmits = {};

socket.on('reconnection', () => resendAllPendingLimits);
socket.on('confirm', (emitID) => { delete(pendingEmits[emitID]); });

// Client
socket.on('something', () => {
    socket.emit('confirm', emitID);
});

解决方案 2(有点)

于 2020 年 2 月 1 日添加。

虽然这不是 Websockets 的真正解决方案，但有人可能仍然觉得它很方便。我们从 Websockets 迁移到 SSE + Ajax。 SSE 允许您从客户端连接以保持持久的 TCP 连接并实时接收来自服务器的消息。要将消息从客户端发送到服务器 - 只需使用 Ajax。存在延迟和开销等缺点，但 SSE 保证可靠性，因为它是 TCP 连接。

由于我们使用 Express，因此我们将此库用于 SSE https://github.com/dpskvn/express-sse ，但您可以选择适合您的那一款。

IE 和大多数 Edge 版本不支持 SSE，因此您需要一个 polyfill:https://github.com/Yaffle/EventSource .

Answer 1

其他人在其他答案和评论中暗示了这一点，但根本问题是 Socket.IO 只是一种交付机制，您不能单独依赖它来实现可靠的交付。唯一能确定消息已成功传递给客户的人是客户自己。对于这种系统，我建议做出以下断言:

1. 消息不会直接发送给客户；相反，它们会被发送到服务器并存储在某种数据存储中。
2. 客户端负责在重新连接时询问“我错过了什么”，并将查询数据存储中存储的消息以更新其状态。
3. 如果在接收方客户端连接时向服务器发送消息，该消息将实时发送到客户端。

当然，根据您的应用程序的需要，您可以调整其中的一部分——例如，您可以使用 Redis 列表或排序集来处理消息，如果您知道一个事实，则将其清除客户端是最新的。

---

这里有几个例子:

幸福之路:

• U1 和 U2 都连接到系统。
• U2 向服务器发送一条 U1 应该接收的消息。
• 服务器将消息存储在某种持久性存储中，用某种时间戳或顺序 ID 为 U1 标记它。
• 服务器通过 Socket.IO 向 U1 发送消息。
• U1 的客户端确认(可能通过 Socket.IO 回调)它收到了消息。
• 服务器从数据存储中删除持久化消息。

离线路径:

• U1 失去互联网连接。
• U2 向服务器发送一条 U1 应该接收的消息。
• 服务器将消息存储在某种持久性存储中，用某种时间戳或顺序 ID 为 U1 标记它。
• 服务器通过 Socket.IO 向 U1 发送消息。
• U1 的客户不确认收货，因为他们处于离线状态。
• 也许 U2 向 U1 发送了更多消息；它们都以相同的方式存储在数据存储中。
• 当 U1 重新连接时，它会询问服务器“我看到的最后一条消息是 X/我有状态 X，我错过了什么。”
• 服务器根据 U1 的请求向 U1 发送它从数据存储中遗漏的所有消息
• U1 的客户端确认收到，服务器从数据存储中删除这些消息。

---

如果您绝对希望有保证的交付，那么重要的是设计您的系统，使连接实际上并不重要，实时交付只是一个奖励；这几乎总是涉及某种数据存储。正如 user568109 在评论中提到的那样，有一些消息传递系统可以抽象出所述消息的存储和传递，并且可能值得研究这种预构建的解决方案。 (您可能仍然需要自己编写 Socket.IO 集成。)

如果您对将消息存储在数据库中不感兴趣，则可以将它们存储在本地数组中；服务器尝试向 U1 发送消息，并将其存储在“待处理消息”列表中，直到 U1 的客户端确认它收到它。如果客户端离线，那么当它回来时，它可以告诉服务器“嘿，我已断开连接，请将我错过的任何内容发送给我”，然后服务器可以遍历这些消息。

幸运的是，Socket.IO 提供了一种机制，允许客户端“响应”类似于原生 JS 回调的消息。这是一些伪代码:

// server
pendingMessagesForSocket = [];

function sendMessage(message) {
  pendingMessagesForSocket.push(message);
  socket.emit('message', message, function() {
    pendingMessagesForSocket.remove(message);
  }
};

socket.on('reconnection', function(lastKnownMessage) {
  // you may want to make sure you resend them in order, or one at a time, etc.
  for (message in pendingMessagesForSocket since lastKnownMessage) {
    socket.emit('message', message, function() {
      pendingMessagesForSocket.remove(message);
    }
  }
});

// client
socket.on('connection', function() {
  if (previouslyConnected) {
    socket.emit('reconnection', lastKnownMessage);
  } else {
    // first connection; any further connections means we disconnected
    previouslyConnected = true;
  }
});

socket.on('message', function(data, callback) {
  // Do something with `data`
  lastKnownMessage = data;
  callback(); // confirm we received the message
});

这与上一个建议非常相似，只是没有持久数据存储。

---

您可能还对 event sourcing 的概念感兴趣.