Socket.io服务端离线事件缓存方案咨询：仅推送离线期间数据变更

针对你遇到的重连全量加载问题，以下是几种实用的实现方案，无需搭建复杂的Socket.io包装层：

方案1：基于用户标识的事件缓存（轻量首选） #

核心思路是按用户ID而非单个Socket缓存离线事件，因为Socket.io服务端确实会销毁断开的Socket实例，但用户标识是持久的。

实现步骤： #

• 客户端连接时传递用户唯一标识（比如从JWT解析的ID、会话ID），服务端将Socket加入对应用户的专属房间。
• 服务端维护一个事件缓存（单实例用Map，分布式场景用Redis），键为用户ID，值为该用户离线期间的MongoDB变更队列。
• 处理MongoDB Change Stream事件时：
  • 用户在线：直接通过用户房间实时推送事件。
  • 用户离线：将事件存入该用户的缓存队列。
• 用户重连时，服务端先推送缓存的离线事件，再清空队列，之后恢复实时推送。

代码示例（服务端）： #

// 单实例用内存缓存，多实例替换为Redis
const userEventCache = new Map();

// 监听MongoDB Change Stream
const changeStream = db.collection('your-collection').watch();
changeStream.on('change', async (change) => {
  const targetUserId = getUserIdFromChange(change); // 从变更数据中关联用户ID
  const userRoom = `user:${targetUserId}`;
  
  // 检查用户是否在线
  if (io.sockets.adapter.rooms.has(userRoom)) {
    io.to(userRoom).emit('data-update', change);
  } else {
    // 缓存离线事件
    if (!userEventCache.has(targetUserId)) {
      userEventCache.set(targetUserId, []);
    }
    userEventCache.get(targetUserId).push(change);
  }
});

// 处理用户连接/重连
io.on('connection', (socket) => {
  const userId = socket.handshake.auth.userId;
  const userRoom = `user:${userId}`;
  socket.join(userRoom);

  // 推送缓存的离线事件
  const cachedEvents = userEventCache.get(userId) || [];
  if (cachedEvents.length > 0) {
    socket.emit('batch-offline-updates', cachedEvents);
    userEventCache.delete(userId);
  }

  // 其他业务逻辑...
});

优缺点： #

• ✅ 轻量易实现，对现有代码侵入小
• ✅ 支持单实例/分布式场景（替换缓存即可）
• ❌ 需要确保变更事件能关联到具体用户；内存缓存重启会丢失事件（分布式场景需用Redis持久化）

方案2：利用MongoDB Change Stream的恢复令牌 #

核心思路是借助MongoDB原生的断点续传能力，客户端记录最后一次接收事件的resumeToken，重连时从该断点拉取离线期间的事件。

实现步骤： #

• 客户端本地存储（如localStorage）最后一次接收事件的resumeToken。
• 用户重连时，将该令牌发送给服务端。
• 服务端根据令牌创建断点续传的Change Stream，拉取离线期间的事件并推送给客户端，之后切换回实时流。

代码示例： #

// 客户端
socket.on('connect', () => {
  const lastResumeToken = localStorage.getItem('lastResumeToken');
  if (lastResumeToken) {
    socket.emit('request-offline-updates', JSON.parse(lastResumeToken));
  }
});

socket.on('data-update', (change) => {
  // 处理数据更新
  localStorage.setItem('lastResumeToken', JSON.stringify(change._id));
});

// 服务端
io.on('connection', (socket) => {
  socket.on('request-offline-updates', async (resumeToken) => {
    // 创建从断点开始的Change Stream
    const resumeStream = db.collection('your-collection').watch([], { resumeAfter: resumeToken });
    
    // 拉取并推送离线事件
    for await (const change of resumeStream) {
      socket.emit('data-update', change);
      // 拉到最新事件后关闭流
      const latestChange = await db.collection('your-collection').findOne({}, { sort: { _id: -1 } });
      if (change._id.toString() === latestChange._id.toString()) {
        resumeStream.close();
        break;
      }
    }
  });
});

优缺点： #

• ✅ 无需服务端缓存事件，利用MongoDB原生能力
• ✅ 适合全局数据（如公告）或用户关联不明确的场景
• ❌ 依赖MongoDB oplog的保留时长（需配置足够大的oplog避免旧数据被覆盖）

关于是否需要搭建缓存Socket的包装层 #

不需要。上述两种方案都是轻量的增量改造，无需对Socket.io核心做包装。只有当你需要通用化的事件缓存机制（比如支持任意事件的离线重发）时，才考虑扩展Socket.io的Adapter或搭建包装层，但这会增加开发和维护成本，优先推荐前两种方案。

内容的提问来源于stack exchange，提问作者istvan kolkert