HTTP 流式接口详解与封装实践

前言

在现代 Web 应用中，流式接口越来越常见，特别是在 AI 对话、实时数据推送等场景中。本文将深入探讨 HTTP 流式接口的工作原理，常见的类型定义问题，以及如何使用 XMLHttpRequest 封装一个通用的流式请求处理方法。

什么是流式接口？

流式接口（Streaming API）是一种服务器可以持续向客户端发送数据的 HTTP 通信方式，而不需要等待所有数据准备完毕。常见的实现方式包括：

  • Server-Sent Events (SSE)：基于 HTTP 的单向通信，使用 text/event-stream 格式

  • WebSocket：全双工通信，适用于双向实时交互

  • Chunked Transfer Encoding：HTTP 分块传输，适用于大文件或长时间数据流

XMLHttpRequest 的流式数据处理机制

关键概念：responseText 的累积特性

很多开发者不理解的一个重要概念是：XMLHttpRequest.responseText 是累积的。

// 错误理解：认为每次onprogress只包含新数据
xhr.onprogress = () => {
  const data = xhr.responseText; // ❌ 这包含从开始到现在的所有数据
};

// 正确理解：需要处理累积数据
let responseText = "";
xhr.onprogress = () => {
  const newText = xhr.responseText.substring(responseText.length); // ✅ 只获取新增数据
  responseText = xhr.responseText; // 更新记录位置
};

onprogress 触发机制

xhr.onprogress 的触发不是按消息触发，而是按数据块（chunk）触发：

  • 触发时机：每当接收到一定量的数据块时就会触发

  • 数据内容：单次事件中可能包含多条完整消息或不完整消息

  • 累积特性：xhr.responseText 始终包含从开始到现在的所有数据

数据流示例

假设服务器分 3 次发送数据：

第1次 onprogress:
xhr.responseText = "data: hello\n\n"
newText = "data: hello\n\n"

第2次 onprogress:
xhr.responseText = "data: hello\n\ndata: world\n\n"  // 包含之前所有数据
newText = "data: world\n\n"  // 只处理新增部分

第3次 onprogress:
xhr.responseText = "data: hello\n\ndata: world\n\ndata: !\n\n"
newText = "data: !\n\n"  // 只处理最新部分

为什么浏览器开发者工具显示不同？

浏览器的 EventStream 面板显示的是"逻辑事件"，它会自动解析 SSE 格式并美化显示：

EventStream 面板显示：
├── data: hello
├── data: world
└── data: !

而你的代码接收的是"原始数据流"，需要手动处理累积数据。这种差异是因为：

  • 开发者工具：自动解析和美化 SSE 格式

  • XMLHttpRequest：提供原始的累积数据流

  • 两者本质：看到的是同一个数据流的不同视角

通用流式请求封装

基于以上分析，我们可以封装一个通用的、生产级别的流式请求处理方法：

interface StreamHttpRequestConfig {
  url: string;
  method: "GET" | "POST";
  headers?: Record<string, string>;
  body?: any;
  params?: Record<string, string>;
  onProgress: (chunk: string) => void; // 处理每个流式数据块
  onComplete?: () => void; // 流结束时的回调
  onError?: (error: Error) => void; // 错误处理回调
  timeout?: number; // 超时时间，默认60秒
}

export const streamHttpRequest = ({
  url,
  method,
  headers = {},
  body,
  params = {},
  onProgress,
  onComplete,
  onError,
  timeout = 60000,
}: StreamHttpRequestConfig): void => {
  const xhr = new XMLHttpRequest();

  // 构建URL参数
  const urlParams = new URLSearchParams(params).toString();
  const finalUrl = urlParams ? `${url}?${urlParams}` : url;

  xhr.open(method, finalUrl, true);

  // 设置流式请求必需的请求头
  xhr.setRequestHeader("Accept", "text/event-stream");
  xhr.setRequestHeader("Cache-Control", "no-cache");

  // 设置自定义请求头
  for (const key in headers) {
    if (headers.hasOwnProperty(key)) {
      xhr.setRequestHeader(key, headers[key]);
    }
  }

  // POST 请求需要设置 Content-Type
  if (method === "POST") {
    xhr.setRequestHeader("Content-Type", "application/json");
  }

  let responseText = ""; // 记录上次处理位置
  let hasError = false;

  xhr.onreadystatechange = () => {
    // 检查响应头接收状态
    if (xhr.readyState === XMLHttpRequest.HEADERS_RECEIVED) {
      if (xhr.status !== 200) {
        hasError = true;
        const errorMsg = `HTTP error! status: ${xhr.status}`;
        onError?.(new Error(errorMsg));
        return;
      }

      // 可选：检查 Content-Type 确保是流式响应
      const contentType = xhr.getResponseHeader("Content-Type") || "";
      console.log("Content-Type:", contentType);
    }
  };

  xhr.onprogress = (_event: ProgressEvent) => {
    // 早期错误检查
    if (hasError) return;

    // 再次检查状态码（防止状态变化）
    if (xhr.status !== 200) {
      hasError = true;
      let errorMessage = `HTTP error! status: ${xhr.status}`;

      try {
        const errorResponse = JSON.parse(xhr.responseText);
        errorMessage =
          errorResponse.message || errorResponse.error || errorMessage;
      } catch {
        errorMessage = xhr.responseText || errorMessage;
      }

      onError?.(new Error(errorMessage));
      return;
    }

    // 调试日志（生产环境可移除）
    console.log("xhr.responseText", xhr.responseText);

    // 关键：获取新增数据块
    const newText = xhr.responseText.substring(responseText.length);
    responseText = xhr.responseText;

    if (newText) {
      // 按行分割并过滤空行
      const lines = newText.split("\n").filter((line) => line.trim());

      for (const line of lines) {
        if (line.startsWith("data:")) {
          const content = line.substring(5).trim();

          // 跳过空数据和结束标记
          if (!content || content === "[DONE]" || content === "") {
            continue;
          }

          try {
            // 尝试解析 JSON 格式的数据
            const parsed = JSON.parse(content);
            const textContent =
              parsed.content || parsed.text || parsed.delta?.content || "";

            if (textContent) {
              onProgress(textContent);
            }
          } catch {
            // JSON 解析失败，直接使用原始内容
            onProgress(content);
          }
        } else if (
          line.trim() &&
          !line.startsWith("event:") &&
          !line.startsWith("id:")
        ) {
          // 处理非标准格式的数据行

          // 检查是否是错误响应
          if (
            line.includes("error") ||
            line.includes("Error") ||
            line.includes("exception")
          ) {
            onError?.(new Error(line.trim()));
            return;
          }

          onProgress(line.trim());
        }
      }
    }
  };

  xhr.onload = () => {
    if (hasError) return;

    if (xhr.status >= 200 && xhr.status < 300) {
      onComplete?.();
    } else {
      // 处理最终错误响应
      let errorMessage = `HTTP error! status: ${xhr.status}`;

      try {
        const errorResponse = JSON.parse(xhr.responseText);
        errorMessage =
          errorResponse.message || errorResponse.error || errorMessage;
      } catch {
        errorMessage = xhr.responseText || errorMessage;
      }

      onError?.(new Error(errorMessage));
    }
  };

  xhr.onerror = () => {
    onError?.(new Error("Network error"));
  };

  xhr.ontimeout = () => {
    onError?.(new Error("Request timeout"));
  };

  // 设置超时时间
  xhr.timeout = timeout;

  xhr.send(method === "POST" ? JSON.stringify(body) : null);
};

封装设计的亮点

1. 类型安全

  • 使用 TypeScript 接口确保参数类型正确

  • 可选参数使用 ? 标记，提高 API 灵活性

2. 错误处理完善

  • 多层次错误检查：状态码、网络错误、超时

  • 智能错误消息解析：尝试从响应中提取具体错误信息

  • 防御性编程：使用 hasError 标志避免重复错误处理

3. 请求头设置正确

  • Accept: text/event-stream：告诉服务器期望 SSE 格式

  • Cache-Control: no-cache：防止缓存影响流式数据

  • 灵活的自定义请求头支持

4. 数据解析健壮

  • 支持标准 SSE 格式（data: 前缀）

  • 兼容非标准格式的流式数据

  • JSON 解析失败时的降级处理

5. 性能优化

  • 使用 substring 只处理新增数据，避免重复处理

  • 早期返回机制减少不必要的计算

  • 内存友好的增量处理方式

最佳实践与优化建议

1. 错误处理策略

// 建议实现重试机制
const retryStreamRequest = (
  config: StreamHttpRequestConfig,
  maxRetries = 3
) => {
  let retryCount = 0;

  const executeRequest = () => {
    streamHttpRequest({
      ...config,
      onError: (error) => {
        if (retryCount < maxRetries) {
          retryCount++;
          console.log(`重试第 ${retryCount} 次...`);
          setTimeout(executeRequest, 1000 * retryCount); // 指数退避
        } else {
          config.onError?.(error);
        }
      },
    });
  };

  executeRequest();
};

2. 内存管理

// 对于长时间运行的流式连接，考虑定期清理
let responseText = "";
const MAX_RESPONSE_SIZE = 1024 * 1024; // 1MB

xhr.onprogress = () => {
  // ... 处理逻辑 ...

  // 防止内存泄漏
  if (responseText.length > MAX_RESPONSE_SIZE) {
    console.warn("响应数据过大，建议检查流式数据处理逻辑");
  }
};

3. 调试和监控

// 生产环境的监控封装
export const streamHttpRequestWithMetrics = (
  config: StreamHttpRequestConfig
) => {
  const startTime = Date.now();
  let chunkCount = 0;

  return streamHttpRequest({
    ...config,
    onProgress: (chunk) => {
      chunkCount++;
      config.onProgress(chunk);
    },
    onComplete: () => {
      const duration = Date.now() - startTime;
      console.log(`流式请求完成: 耗时${duration}ms, 接收${chunkCount}个数据块`);
      config.onComplete?.();
    },
  });
};

4. 类型安全增强

// 使用泛型增强类型安全
interface StreamResponse<T> {
  content?: string;
  data?: T;
  error?: string;
}

interface StreamHttpRequestConfig<T = any> {
  // ... 其他属性
  onProgress: (chunk: string, parsed?: StreamResponse<T>) => void;
}

总结

HTTP 流式接口的实现需要深入理解底层机制：

  1. 核心理念：理解 XMLHttpRequest.responseText 的累积特性是关键

  2. 类型安全：正确处理 TypeScript 类型定义，避免参数名冲突

  3. 错误处理：实现多层次、健壮的错误处理机制

  4. 性能优化：使用增量处理避免重复计算和内存泄漏

  5. 生产就绪：考虑重试、监控、调试等生产环境需求

通过封装通用的处理方法，我们可以构建出稳定、可维护的流式数据处理方案，为用户提供流畅的实时交互体验。

希望这篇文章能帮助你深入理解 HTTP 流式接口的实现原理和最佳实践！

最后更新时间:
贡献者: 何风顺