一尘不染

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

node.js

用过的

NodeJS,Socket.io

问题

想象有两个用户 U1U2 通过Socket.io连接到应用程序。该算法如下:

  1. U1 完全失去Internet连接(例如,关闭Internet)
  2. U2U1 发送消息。
  3. U1 尚未收到消息,因为Internet断开
  4. 服务器 通过心跳超时检测到 U1 断开连接
  5. U1 重新连接到socket.io
  6. U1 从不接收来自 U2 的消息-我猜它在步骤4中丢失了。

可能的解释

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

  • 第4步 服务器 杀死Socket实例和消息队列 U1 以及
  • 此外,在步骤5中, U1服务器 创建新的连接(不重用),因此即使消息仍在排队中,先前的连接还是会丢失。

需要帮忙

如何防止这种数据丢失?我必须使用心跳,因为我不会有人永远挂在应用程序中。另外,我仍然必须提供重新连接的可能性,因为当我部署新版本的应用程序时,我希望停机时间为零。

PS我称之为“消息”的东西不仅是我可以存储在数据库中的文本消息,而且是有价值的系统消息,必须保证其传递或UI搞砸。

谢谢!


加法1

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

签出第2步。从技术上讲,在这一步上我们 不能说U1是否脱机
,他只是失去了连接状态,说了2秒钟,这可能是因为互联网状况不佳。因此,U2向他发送了一条消息,但是U1没有收到该消息,因为互联网对他来说仍然不可用(步骤3)。需要步骤4来检测脱机用户,可以说超时是60秒。最终在另外10秒内,U1的互联网连接建立,他重新连接到socket.io。但是来自U2的消息在空间中丢失了,因为服务器U1上的超时已断开连接。

那就是问题,我不会100%交货。


  1. 收集{}用户中的发射(发射名称和数据),由随机的emitID标识。发送发射
  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。

阅读 997

收藏
2020-07-07

共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
});

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


您可能还对事件源的概念感兴趣。

2020-07-07