WebSocket循环调用接口获取用户功能失效问题求助

核心问题 #

你的代码存在三个关键问题导致程序卡住：

1. 双重JSON序列化错误：你将已经是JSON格式的转义字符串再次用JSON.stringify包裹，导致服务端收到嵌套的JSON字符串，无法正确解析后续请求。
2. Promise未处理异常场景：fetchUser的Promise仅在收到正确结果时resolve，但连接出错、提前关闭或解析失败时，Promise会一直处于pending状态，导致await永远卡住，后续循环无法执行。
3. 请求ID重复：所有请求固定使用ID "1"，服务端无法区分不同请求的响应，可能导致后续请求的响应无法被正确识别。

修复后的单用户请求代码 #

import generateConnection from './generate-connection';

export async function fetchUser(id: number) {
  return new Promise((resolve, reject) => {
    const connection = generateConnection();
    let hasResolved = false;

    // 清理连接并处理未完成的Promise
    const cleanup = () => {
      if (!hasResolved) {
        reject(new Error(`连接提前关闭，未获取用户${id}的信息`));
      }
      connection.close();
    };

    connection.onopen = () => {
      // 直接发送JSON对象，避免双重序列化
      connection.send(JSON.stringify({
        msg: "connect",
        version: "1",
        support: ["1", "pre2", "pre1"]
      }));
      // 用用户ID作为请求ID，确保唯一性
      connection.send(JSON.stringify({
        msg: "method",
        id: String(id),
        method: "Users.getUser",
        params: [id]
      }));
      console.log(`已连接，开始获取用户${id}`);
    };

    connection.on('message', (event) => {
      const data = event.toString();
      if (data[0] !== 'a') return;

      try {
        const parsedData = JSON.parse(data.substring(1));
        const response = JSON.parse(parsedData[0]);
        
        if (response.msg === 'result') {
          hasResolved = true;
          if ('error' in response) {
            reject(new Error(`获取用户${id}失败：${response.error.msg}`));
          } else {
            resolve(response.result);
          }
          cleanup(); // 获取结果后主动关闭连接
        }
      } catch (parseError) {
        hasResolved = true;
        reject(new Error(`解析用户${id}响应失败：${parseError.message}`));
        cleanup();
      }
    });

    connection.on('error', (error) => {
      hasResolved = true;
      reject(new Error(`用户${id}连接出错：${error.toString()}`));
      cleanup();
    });

    connection.on('close', cleanup);
  });
}

const fetchAllUsers = async () => {
  let successCount = 0;
  for (let i = 0; i < 100; i++) {
    try {
      const user = await fetchUser(i);
      console.log(`用户${i}：${user.name}`);
      successCount++;
      // 这里添加存入数据库的逻辑，例如：
      // await saveUserToDatabase(user);
    } catch (err) {
      console.error(`获取用户${i}失败：`, err.message);
      // 可选：遇到错误时是否继续执行，取消注释下方代码即可
      // continue;
    }
    // 可选：添加延迟避免频繁创建连接触发服务端限流
    // await new Promise(resolve => setTimeout(resolve, 100));
  }
  console.log(`完成：成功获取${successCount}个用户`);
};

fetchAllUsers();

高效批量处理方案（适合5000用户场景） #

每次新建连接效率极低，推荐复用单个连接批量发送请求，大幅减少连接开销：

import generateConnection from './generate-connection';

// 批量获取用户
export async function fetchUserBatch(ids: number[]) {
  return new Promise((resolve, reject) => {
    const connection = generateConnection();
    const results = new Map<number, any>();
    const errors = new Map<number, Error>();
    const expectedCount = ids.length;

    // 检查是否所有请求都已处理
    const checkCompletion = () => {
      if (results.size + errors.size === expectedCount) {
        connection.close();
        resolve({
          success: Array.from(results.entries()).map(([id, user]) => ({ id, user })),
          failed: Array.from(errors.entries()).map(([id, err]) => ({ id, error: err.message }))
        });
      }
    };

    connection.onopen = () => {
      // 发送连接初始化消息
      connection.send(JSON.stringify({
        msg: "connect",
        version: "1",
        support: ["1", "pre2", "pre1"]
      }));
      // 批量发送所有用户请求
      ids.forEach(id => {
        connection.send(JSON.stringify({
          msg: "method",
          id: String(id),
          method: "Users.getUser",
          params: [id]
        }));
      });
      console.log(`已连接，开始批量处理${expectedCount}个用户`);
    };

    connection.on('message', (event) => {
      const data = event.toString();
      if (data[0] !== 'a') return;

      try {
        const parsedData = JSON.parse(data.substring(1));
        const response = JSON.parse(parsedData[0]);
        
        if (response.msg === 'result') {
          const userId = parseInt(response.id);
          if ('error' in response) {
            errors.set(userId, new Error(response.error.msg));
          } else {
            results.set(userId, response.result);
          }
          checkCompletion();
        }
      } catch (parseError) {
        console.error(`响应解析失败：${parseError.message}`);
      }
    });

    connection.on('error', (error) => {
      reject(new Error(`批量请求连接出错：${error.toString()}`));
      connection.close();
    });

    connection.on('close', () => {
      if (results.size + errors.size < expectedCount) {
        reject(new Error(`连接提前关闭，未完成全部${expectedCount}个请求`));
      }
    });
  });
}

// 分批处理5000个用户
const fetchAllUsers = async () => {
  const totalUsers = 5000;
  const batchSize = 100; // 每次批量处理100个，可根据服务端调整
  let totalSuccess = 0;

  for (let i = 0; i < totalUsers; i += batchSize) {
    const batchIds = Array.from({ length: Math.min(batchSize, totalUsers - i) }, (_, idx) => i + idx);
    try {
      const { success, failed } = await fetchUserBatch(batchIds);
      totalSuccess += success.length;
      console.log(`批次${Math.floor(i/batchSize)+1}：成功${success.length}个，失败${failed.length}个`);
      
      // 批量存入数据库
      // await Promise.all(success.map(({ user }) => saveUserToDatabase(user)));
      
      // 输出失败的用户ID
      failed.forEach(({ id, error }) => console.error(`用户${id}失败：${error}`));
    } catch (err) {
      console.error(`批次${Math.floor(i/batchSize)+1}整体失败：`, err.message);
    }
    // 批次间添加延迟，避免服务端压力过大
    await new Promise(resolve => setTimeout(resolve, 500));
  }

  console.log(`全部完成：成功获取${totalSuccess}/${totalUsers}个用户`);
};

fetchAllUsers();

其他可行方案 #

如果服务端支持HTTP接口，优先使用HTTP批量请求（例如POST /api/users/batch传入ID列表），WebSocket更适合实时双向通信，批量数据同步场景下HTTP实现更简单，也更容易处理重试、限流等问题。

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