全面掌握浏览器流式请求：从 XMLHttpRequest 到 fetchEventSource

全面掌握浏览器流式请求：从 XMLHttpRequest 到 fetchEventSource

• 引言：为什么需要流式数据传输？
• 什么是流式数据传输？
• • 传统请求 vs 流式请求的区别
  • 常见应用场景：
• 基础篇：初识流式请求
• • 最传统的方式：XMLHttpRequest
  • • XHR基本使用示例
    • XHR 的流式特性
    • XHR 流式的优点：
    • XHR 流式的缺点：
  • 浏览器原生方案：EventSource
  • • 什么是 SSE？
    • EventSource 基本使用
    • EventSource 的优势
    • EventSource 的限制
• 进阶篇：现代流式请求方案
• • Fetch API + ReadableStream
  • • 基础流式读取
  • 终极方案：fetchEventSource
  • • 基础用法
    • 为什么选择 fetchEventSource？
• 深度对比：五种方案全面解析
• 底层原理深度解析
• • HTTP 连接的本质
  • • HTTP/1.1 的连接限制
    • HTTP/2 的多路复用
  • EventSource 的自动重连机制
  • fetchEventSource 的实现原理
  • • fetchEventSource 的核心架构
    • fetchEventSource 的关键设计要点
  • AbortController 的工作原理
  • • AbortController 的底层实现
    • 为什么 AbortController 能中止 fetchEventSource？
• 未来趋势
• 常见问题解答
• • 流式请求会占用太多连接吗？
  • 如何保证数据完整性？
  • 流式请求会影响页面性能吗？
  • 如何处理网络不稳定？
• 结语

引言：为什么需要流式数据传输？

想象一下这样的场景：你正在使用一个实时股票交易网站，股价每秒钟都在变化；或者使用 ChatGPT，看着它一个字一个字地"思考"和回答。这些体验的背后，都离不开一项关键技术：流式数据传输。

什么是流式数据传输？

简单来说，传统的数据请求就像寄送一个包裹：我们在下单之后，等待，然后一次性收到完整的包裹；而流式数据传输就像打开水龙头：水（数据）源源不断地流出来，我们可以边接收边处理。

传统请求 vs 流式请求的区别

// 传统方式：一次性接收
fetch('/api/data')
.then(res => res.json()) // 等待所有数据
.then(data => process(data)); // 然后处理

// 流式方式：边接收边处理
fetch('/api/stream')
.then(res => {
const reader = res.body.getReader(); // 立即获取流
// 数据到达一块就处理一块
return reader.read().then(processChunk);
});

常见应用场景：

• 实时聊天应用
• 股票行情更新
• AI 对话的逐字显示
• 大文件下载的进度显示
• 实时日志监控

在本文中，我们将从最古老的 XMLHttpRequest 开始，一直到现代流行的 fetchEventSource，全面掌握浏览器中的流式请求技术。

基础篇：初识流式请求

最传统的方式：XMLHttpRequest

如果我们使用过 AJAX，那就应该清楚它的核心就是 XMLHttpRequest（简称 XHR）。虽然名字里有 “XML”，但它却可以处理任何类型的数据。

XHR基本使用示例

// 创建一个 XHR 对象
const xhr = new XMLHttpRequest();
// 配置请求
xhr.open('GET', '/api/data');
// 监听状态变化
xhr.onreadystatechange = function() {
// readyState 有 5 个状态：
// 0: 未初始化, 1: 已打开, 2: 已发送, 3: 接收中, 4: 完成
if (xhr.readyState === 4 && xhr.status === 200) {
console.log('完整数据:', xhr.responseText);
}
};
// 发送请求
xhr.send();

XHR 的流式特性

很多人不知道，XHR 其实也支持"流式"接收数据：当 readyState 为 3 时，表示数据正在接收中，responseText 就包含已接收的部分数据：

const xhr = new XMLHttpRequest();
xhr.open('GET', '/api/stream-data');

let receivedLength = 0;

xhr.onreadystatechange = function() {
// 关键：当 readyState 为 3 时，数据正在接收中
if (xhr.readyState === 3) {
// responseText 包含已接收的部分数据
const currentData = xhr.responseText;

// 只处理新到达的数据
const newData = currentData.slice(receivedLength);
receivedLength = currentData.length;

if (newData) {
console.log('新数据块:', newData);
// 这里可以实时处理数据
}
}

if (xhr.readyState === 4) {
console.log('数据传输完成');
}
};

xhr.send();

XHR 流式的优点：

• 浏览器支持极好（包括 IE）
• 无需额外库
• 可以监控下载进度

XHR 流式的缺点：

• API 设计老旧
• 无法精确控制数据块边界
• 代码可读性差

浏览器原生方案：EventSource

HTML5 引入了专门的服务器推送技术：EventSource，专门用于处理 Server-Sent Events（SSE）。

什么是 SSE？

SSE 是一种允许服务器向客户端推送数据的技术。与 WebSocket 不同，SSE 是单向的（只能服务器→客户端），基于普通的 HTTP 协议。

对于SSE 与 WebSocket 的对比，在我的另一篇文章中有详细讲解，本文不再赘述：SSE vs WebSocket：实时通信技术全面对比

EventSource 基本使用

// 创建 EventSource 连接
const eventSource = new EventSource('/api/events');

// 监听消息事件
eventSource.onmessage = function(event) {
console.log('收到消息:', event.data);
};

// 监听自定义事件
eventSource.addEventListener('update', function(event) {
const data = JSON.parse(event.data);
console.log('更新事件:', data);
});

// 错误处理
eventSource.onerror = function(error) {
console.error('连接错误:', error);
// EventSource 会自动尝试重连
};

// 关闭连接（当不再需要时）
eventSource.close();

EventSource 的优势

• 自动重连机制
• 简单易用的 API
• 内置事件解析

EventSource 的限制

• 只能使用 GET 请求
• 不能自定义请求头
• 不支持发送数据到服务器
• 部分浏览器不支持（如 IE）

进阶篇：现代流式请求方案

Fetch API + ReadableStream

Fetch API 是现代浏览器提供的更强大的网络请求接口，结合 ReadableStream 可以实现灵活的流式处理。

基础流式读取

// 发起 fetch 请求
fetch('/api/stream')
.then(response => {
// 获取可读流
const reader = response.body.getReader();
const decoder = new TextDecoder('utf-8');

function readChunk() {
return reader.read().then(({ done, value }) => {
if (done) {
console.log('流读取完成');
return;
}
// 将二进制数据解码为文本
const chunk = decoder.decode(value);
console.log('收到数据块:', chunk);

// 继续读取下一块
return readChunk();
});
}
return readChunk();
})
.catch(error => {
console.error('请求失败:', error);
});

上述代码可以改成 async/await 更优雅的写法：

async function fetchStream(url) {
try {
const response = await fetch(url);
const reader = response.body.getReader();
const decoder = new TextDecoder();

while (true) {
const { done, value } = await reader.read();
if (done) {
console.log('流结束');
break;
}
const text = decoder.decode(value);
console.log('接收数据:', text);
// 在这里处理数据
processData(text);
}
} catch (error) {
console.error('错误:', error);
}
}

终极方案：fetchEventSource

fetchEventSource 结合了 Fetch API 的灵活性和 SSE 的便利性，是目前最推荐的方案。

基础用法

import { fetchEventSource } from '@microsoft/fetch-event-source';

// 基础示例
await fetchEventSource('/api/chat-stream', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': 'Bearer your-token-here'
},
body: JSON.stringify({
message: '你好，请介绍一下你自己'
}),
onopen(response) {
// 连接成功时调用
console.log('连接已建立，状态码:', response.status);
if (response.ok) {
console.log('连接正常');
} else {
throw new Error('连接失败');
}
},
onmessage(event) {
// 每次收到消息时调用
const data = event.data;
console.log('收到消息:', data);

// 如果是 JSON 数据
try {
const parsed = JSON.parse(data);
updateUI(parsed);
} catch (e) {
// 直接文本数据
appendText(data);
}
},
onerror(err) {
// 发生错误时调用
console.error('流错误:', err);
// 注意：如果抛出错误，会自动重试
// 如果不想重试，不要抛出错误
},
onclose() {
// 连接关闭时调用
console.log('连接已关闭');
}
});

为什么选择 fetchEventSource？

• 灵活的请求配置：可以使用任何 HTTP 方法、自定义头部
• 完整的错误控制：可以自定义重试逻辑
• SSE 协议支持：自动解析 SSE 格式数据
• 现代 API 设计：基于 Promise，支持 async/await

深度对比：五种方案全面解析

底层原理深度解析

HTTP 连接的本质

HTTP/1.1 的连接限制

浏览器对同一域名的并发连接限制

• Chrome/Firefox: 6个连接
• IE 7/8: 2个连接（重要！）
• IE 9/10: 6个连接
• IE 11: 8个连接

这意味着：如果使用了6个EventSource连接，那页面上的其他资源请求会被阻塞！

HTTP/2 的多路复用

HTTP/2 的多路复用技术，虽然解决了连接限制问题，但仍需注意：

EventSource 的自动重连机制

其核心处理逻辑如下：

class EventSourceSimulator {
constructor(url) {
this.url = url;
this.reconnectInterval = 3000; // 默认3秒
this.lastEventId = '';
this.isConnecting = false;

this.connect();
}

connect() {
if (this.isConnecting) return;

this.isConnecting = true;
const xhr = new XMLHttpRequest();

// 设置请求头
xhr.open('GET', this.url);
xhr.setRequestHeader('Accept', 'text/event-stream');
xhr.setRequestHeader('Cache-Control', 'no-cache');

if (this.lastEventId) {
xhr.setRequestHeader('Last-Event-ID', this.lastEventId);
}

xhr.onreadystatechange = () => {
if (xhr.readyState === 3) {
// 解析 SSE 数据流
}

if (xhr.readyState === 4) {
this.isConnecting = false;

// 服务器可以通过 retry 字段控制重连时间
if (xhr.status === 200) {
// 正常关闭，不重连
} else {
// 异常断开，等待重连
setTimeout(() => this.connect(), this.reconnectInterval);
}
}
};

xhr.send();
}
}

fetchEventSource 的实现原理

fetchEventSource 的核心架构

async function simpleFetchEventSource(url, options) {
const {
signal, // AbortSignal
onopen, // 连接打开回调
onmessage, // 消息回调
onerror, // 错误回调
onclose, // 关闭回调
fetch, // fetch 函数
openWhenHidden, // 页面隐藏时是否保持连接
…fetchOptions // fetch 的其他参数
} = options;

// 1. 准备请求
const request = {
…fetchOptions,
headers: {
'Accept': 'text/event-stream',
'Cache-Control': 'no-cache',
…fetchOptions.headers,
},
signal, // 关键：传递 AbortSignal
};

// 2. 发起请求
const response = await fetch(url, request);

// 3. 连接建立回调
await (onopen && onopen(response));

// 4. 检查响应状态
if (!response.ok) {
throw new Error(`HTTP ${response.status}`);
}

// 5. 获取可读流
const reader = response.body.getReader();
const decoder = new TextDecoder();

// 6. SSE 解析状态机
let buffer = '';
let eventName = 'message';
let dataBuffer = [];
let currentId = null;
let retryTime = null;

try {
// 7. 主循环：读取流数据
while (true) {
// 8. 检查是否被中止
if (signal && signal.aborted) {
throw new DOMException('Aborted', 'AbortError');
}

// 读取数据块
const { done, value } = await reader.read();

if (done) {
// 流结束
break;
}

// 9. 解码并处理数据块
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split('\\n');
// 最后一行可能不完整，保留在 buffer 中
buffer = lines.pop() || '';

for (let line of lines) {
line = line.trimEnd(); // 移除行尾空格

if (line === '') {
// 空行：事件结束，触发回调
if (dataBuffer.length > 0) {
const event = {
id: currentId,
data: dataBuffer.join('\\n'),
event: eventName,
};

if (onmessage) {
onmessage(event);
}
}

// 重置状态
dataBuffer = [];
eventName = 'message';
} else if (line.startsWith(':')) {
// 注释行，忽略
continue;
} else if (line.startsWith('event:')) {
// 事件类型
eventName = line.substring(6).trim();
} else if (line.startsWith('data:')) {
// 数据行
dataBuffer.push(line.substring(5).trim());
} else if (line.startsWith('id:')) {
// 事件ID
currentId = line.substring(3).trim();
} else if (line.startsWith('retry:')) {
// 重连时间
retryTime = parseInt(line.substring(6).trim(), 10);
if (!isNaN(retryTime)) {
// 可以在这里更新重连策略
}
}
// 其他字段可以在这里扩展
}
}
} catch (error) {
// 10. 错误处理
if (error.name === 'AbortError') {
// 用户主动中止，不触发错误回调
if (onclose) onclose();
return;
}

if (onerror) {
onerror(error);
}

// 实现重试逻辑
await retryConnection();
} finally {
// 11. 清理资源
if (onclose) onclose();
reader.releaseLock();
}
}

fetchEventSource 的关键设计要点

• 基于 Fetch API：利用现代浏览器的 fetch 能力
• 完整的 SSE 解析：实现完整的 Server-Sent Events 协议
• 错误恢复机制：智能重试策略
• 资源管理：正确的流关闭和清理
• 信号支持：与 AbortController 深度集成

AbortController 的工作原理

AbortController 的底层实现

class SimpleAbortController {
constructor() {
this.signal = new SimpleAbortSignal();
}

abort(reason) {
this.signal.abort(reason);
}
}

class SimpleAbortSignal extends EventTarget {
constructor() {
super();
this.aborted = false;
this.reason = undefined;
}

abort(reason) {
if (this.aborted) return;

this.aborted = true;
this.reason = reason || new DOMException('Aborted', 'AbortError');

// 触发 abort 事件
this.dispatchEvent(new Event('abort'));
}

// 静态方法：创建已中止的信号
static abort(reason) {
const signal = new SimpleAbortSignal();
signal.abort(reason);
return signal;
}
}

为什么 AbortController 能中止 fetchEventSource？

• fetchEventSource 底层使用的仍然是 fetch API ，在 fetch 请求中传递 signal
• 在流读取循环中，会通过 signal 检查中止状态

async function fetchEventSourceWithAbort(url, options) {
const controller = options.signal?._controller || new AbortController();
const signal = controller.signal;

// 1. 在 fetch 请求中传递 signal
const response = await fetch(url, {
…options,
signal // fetch API 原生支持 AbortSignal
});

// 2. 在流读取循环中检查中止状态
const reader = response.body.getReader();

// 创建一个包装的 read 方法
const abortableRead = async () => {
// 每次读取前检查
signal.throwIfAborted();

try {
return await reader.read();
} catch (error) {
// 如果读取过程中被中止
if (signal.aborted) {
throw new DOMException('Aborted', 'AbortError');
}
throw error;
}
};

// 3. 监听中止事件
const onAbort = () => {
// 中止读取器
reader.cancel();

// 清理资源
cleanupResources();
};

signal.addEventListener('abort', onAbort);

try {
// 主循环
while (true) {
const { done, value } = await abortableRead();
// … 处理数据
}
} catch (error) {
if (error.name === 'AbortError') {
// 用户中止，正常退出
return;
}
throw error;
} finally {
signal.removeEventListener('abort', onAbort);
}
}

未来趋势

• HTTP/3 的 QUIC 协议：更快的连接建立和更好的多路复用
• WebTransport API：基于 QUIC 的现代传输协议
• Service Worker 流式缓存：离线可用的流式数据
• Edge Computing：在边缘节点处理流式数据

常见问题解答

流式请求会占用太多连接吗？

一个 SSE 连接只是一个持久的 HTTP 连接，现代浏览器通常支持 6-8 个并发连接。对于大多数应用来说足够了。

如何保证数据完整性？

SSE 协议本身有重连机制，重连时会发送最后一个事件的 ID，服务器可以从该点继续发送。

流式请求会影响页面性能吗？

流式请求在合理使用时不会影响页面性能，但要注意：

• 及时清理不需要的连接
• 避免同时打开太多流
• 使用 AbortController 管理生命周期

如何处理网络不稳定？

实现指数退避重连：

let retryDelay = 1000; // 初始1秒

function connectWithBackoff() {
connect().catch((error) => {
console.log(`连接失败，${retryDelay}ms后重试`);

setTimeout(() => {
retryDelay *= 2; // 每次失败后加倍等待时间
if (retryDelay > 30000) retryDelay = 30000; // 最大30秒
connectWithBackoff();
}, retryDelay);
});
}

结语

流式请求是现代 Web 应用的核心技术之一，理解其底层原理和不同方案的优缺点，能够帮助我们在实际项目中做出正确的技术决策，对于文章中错误的地方或者有任何问题，欢迎在评论区留言讨论！