在GraphQL Apollo Node.js服务器处理用户上下线事件的最优方案咨询

我之前也碰到过类似的困扰，你的现有方案确实不够优雅，而且依赖订阅的filter逻辑容易出现边界问题（比如客户端异常断开时可能触发不及时）。下面是我实践下来最可靠、清晰的实现方式，核心是利用WebSocket连接的生命周期钩子，直接关联用户身份并处理上下线：

核心思路 #

Apollo的订阅服务器（不管是subscriptions-transport-ws还是官方推荐的graphql-ws）都提供了onConnect和onDisconnect钩子，我们可以在这两个钩子中：

1. 在连接建立时验证用户身份（通过JWT），并把连接ID和用户ID绑定。
2. 在连接断开时，通过连接ID找到对应的用户ID，执行下线逻辑。

这种方式不需要依赖订阅的filter，逻辑和业务订阅完全分离，可靠性更高。

具体实现步骤 #

1. 维护活跃连接映射 #

首先创建一个全局的Map来存储连接ID和用户ID的对应关系，方便断开时快速查找：

// 全局维护活跃连接：key是连接ID，value是用户ID
const activeConnections = new Map();

2. 配置订阅服务器的生命周期钩子 #

这里分两种常见的订阅库来举例：

情况1：使用subscriptions-transport-ws

import { SubscriptionServer } from 'subscriptions-transport-ws';
import { execute, subscribe } from 'graphql';
import { verifyJwtToken } from './utils/auth'; // 替换成你的JWT验证函数

// 假设你已经有了WebSocket服务器实例wsServer和GraphQL schema
const subscriptionServer = SubscriptionServer.create(
  {
    schema,
    execute,
    subscribe,
    // 连接建立时的钩子
    async onConnect(connectionParams, webSocket, context) {
      // 从连接参数或请求头中获取Authorization token
      const authToken = connectionParams.authorization 
        || webSocket.upgradeReq.headers.authorization?.replace('Bearer ', '');
      
      if (!authToken) {
        throw new Error('未授权，请提供有效token'); // 直接拒绝未授权连接
      }

      try {
        // 解析token获取用户信息
        const user = await verifyJwtToken(authToken);
        const connectionId = context.connection.context.connectionId;

        // 绑定连接ID和用户ID
        activeConnections.set(connectionId, user.id);

        // 执行用户上线逻辑：比如更新数据库状态、发布上线事件
        await updateUserOnlineStatus(user.id, true);
        pubSub.publish('USER_ONLINE', { userId: user.id });

        // 返回上下文，后续订阅的resolver可以使用用户信息
        return { userId: user.id };
      } catch (err) {
        throw new Error('无效的token，请重新登录');
      }
    },
    // 连接断开时的钩子
    async onDisconnect(webSocket, context) {
      const connectionId = context.connectionId;
      const userId = activeConnections.get(connectionId);

      if (userId) {
        // 执行用户下线逻辑
        await updateUserOnlineStatus(userId, false);
        pubSub.publish('USER_OFFLINE', { userId: user.id });
        
        // 移除活跃连接记录
        activeConnections.delete(connectionId);
      }
    },
  },
  {
    server: wsServer,
    path: '/subscriptions',
  }
);

情况2：使用graphql-ws（Apollo Server v3+推荐）

如果用的是Apollo Server最新版本，官方推荐用graphql-ws，配置类似但API略有不同：

import { ApolloServer } from '@apollo/server';
import { createServer } from 'http';
import { WebSocketServer } from 'ws';
import { useServer } from 'graphql-ws/lib/use/ws';
import { verifyJwtToken } from './utils/auth';

const httpServer = createServer(app); // 你的Express HTTP服务器
const wsServer = new WebSocketServer({
  server: httpServer,
  path: '/subscriptions',
});

// 配置WebSocket服务器
const serverCleanup = useServer(
  {
    schema,
    // 订阅上下文（可选，和onConnect配合）
    context: async (ctx) => {
      const authToken = ctx.connectionParams?.authorization?.replace('Bearer ', '');
      if (authToken) {
        const user = await verifyJwtToken(authToken);
        return { userId: user.id };
      }
      return {};
    },
    // 连接建立钩子
    async onConnect(ctx) {
      const authToken = ctx.connectionParams?.authorization?.replace('Bearer ', '');
      if (!authToken) {
        throw new Error('未授权连接');
      }

      const user = await verifyJwtToken(authToken);
      activeConnections.set(ctx.connectionId, user.id);
      await updateUserOnlineStatus(user.id, true);
      pubSub.publish('USER_ONLINE', { userId: user.id });

      return { userId: user.id };
    },
    // 连接断开钩子
    async onDisconnect(ctx) {
      const userId = activeConnections.get(ctx.connectionId);
      if (userId) {
        await updateUserOnlineStatus(userId, false);
        pubSub.publish('USER_OFFLINE', { userId: user.id });
        activeConnections.delete(ctx.connectionId);
      }
    },
  },
  wsServer
);

// 记得在服务器关闭时清理WebSocket连接
process.on('SIGTERM', async () => {
  await serverCleanup.dispose();
  httpServer.close();
});

为什么这个方案更优？ #

• 逻辑分离：上下线逻辑和业务订阅完全解耦，不需要通过订阅的filter间接处理，代码更清晰。
• 可靠性高：不管客户端是正常登出还是异常断开（比如网络中断、浏览器崩溃），WebSocket的onDisconnect都会触发，能准确处理下线。
• 身份准确：在连接建立时就验证用户身份，直接绑定连接和用户，不会出现身份混淆的问题。
• 扩展性好：活跃连接的Map还可以用来做其他功能，比如统计在线用户数、定向推送消息等。

注意事项 #

• 一定要在onConnect中验证token，拒绝未授权的连接，避免无效连接占用资源。
• 全局的activeConnections Map在服务器重启时会清空，正常情况下不会有内存泄漏（因为每个断开的连接都会触发onDisconnect删除记录）。
• 如果你的JWT有过期时间，可以考虑在连接期间定期验证，但一般WebSocket连接本身不会长期保持（客户端会在token过期后重新连接）。

内容的提问来源于stack exchange，提问作者Andrew Gura