Loading...

技术

53 分钟

网络请求与通信机制

JavaScript 进阶

Site Owner

Published on 2026-05-21

系统讲解XHR完整状态机、Fetch与Promise集成、Response流式读取、AbortController取消机制及实时通信方案对比

网络请求与通信机制

网络请求与通信机制

现代 Web 应用的网络通信远不止简单的 HTTP 请求。从 Fetch API 的流式响应、AbortController 取消机制，到 WebSocket 全双工通信、Server-Sent Events 推送、WebRTC 点对点连接，本文全面覆盖 JavaScript 网络通信的底层机制与最佳实践。

目录

Fetch API 深度用法
请求拦截与中间件
流式响应处理
WebSocket 全双工通信
Server-Sent Events
请求取消与竞态处理
实战案例
深度追问
总结表格

1. Fetch API 深度用法

// 代码示例 1：Fetch 的完整配置
const response = await fetch('/api/data', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer token123'
  },
  : .({ :  }),
  : .(),  
  : ,
  : ,  
  : ,          
  :           
});


 (!response.)   ();
 data =  response.();


  () {
   totalSize = file.;
   uploadedSize = ;
  
   stream = file.().( ({
    () {
      uploadedSize += chunk.;
      (uploadedSize / totalSize);
      controller.(chunk);
    }
  }));
  
   (url, {
    : ,
    : stream,
    : ,  
    : { : file. }
  });
}

JavaScript网络请求全解—从XHR到Fetch、Streams与WebSocket|新宇宙博客

// 代码示例 3：Fetch 中间件模式
class HttpClient {
  #middlewares = [];
  #baseUrl;
  
  constructor(baseUrl) {
    this.#baseUrl = baseUrl;
  }
  
  use(middleware) {
    this.#middlewares.push(middleware);
    return this;
  }
  
  async request(path, options = {}) {
    const url = new URL(path, this.#baseUrl).href;
    let config = { url, ...options };
    
    // 构建中间件链
    const chain = this.#middlewares.reduceRight(
      (next, middleware) => (cfg) => middleware(cfg, next),
      (cfg) => fetch(cfg.url, cfg).then(this.#handleResponse)
    );
    
    return chain(config);
  }
  
  #handleResponse(response) {
    if (!response.ok) {
      const error = new Error(`HTTP ${response.status}`);
      error.response = response;
      throw error;
    }
    const contentType = response.headers.get('content-type');
    return contentType?.includes('json') ? response.json() : response.text();
  }
}

// 中间件
const authMiddleware = async (config, next) => {
  config.headers = { ...config.headers, Authorization: `Bearer ${getToken()}` };
  return next(config);
};

const retryMiddleware = async (config, next) => {
  for (let i = 0; i < 3; i++) {
    try { return await next(config); }
    catch (e) { if (i === 2) throw e; await sleep(1000 * (i + 1)); }
  }
};

const client = new HttpClient('https://api.example.com')
  .use(authMiddleware)
  .use(retryMiddleware);

// 代码示例 4：SSE 风格的流式 JSON 解析
async function* streamJSON(url) {
  const response = await fetch(url);
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = '';
  
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split('\n');
    buffer = lines.pop(); // 保留不完整的行
    
    for (const line of lines) {
      if (line.trim()) {
        yield JSON.parse(line);
      }
    }
  }
  
  if (buffer.trim()) yield JSON.parse(buffer);
}

// AI 聊天流式响应
for await (const chunk of streamJSON('/api/chat/stream')) {
  appendToUI(chunk.text);
}

// 代码示例 5：健壮的 WebSocket 客户端
class RobustWebSocket {
  #url; #ws; #queue = [];
  #reconnecting = false;
  #handlers = new Map();
  #heartbeatInterval;
  
  constructor(url, options = {}) {
    this.#url = url;
    this.maxReconnectDelay = options.maxReconnectDelay || 30000;
    this.connect();
  }
  
  connect() {
    this.#ws = new WebSocket(this.#url);
    
    this.#ws.onopen = () => {
      this.#reconnecting = false;
      this.#flushQueue();
      this.#startHeartbeat();
    };
    
    this.#ws.onmessage = (e) => {
      const msg = JSON.parse(e.data);
      if (msg.type === 'pong') return;
      this.#handlers.get(msg.type)?.forEach(h => h(msg.data));
    };
    
    this.#ws.onclose = (e) => {
      this.#stopHeartbeat();
      if (!e.wasClean) this.#reconnect();
    };
  }
  
  send(type, data) {
    const message = JSON.stringify({ type, data });
    if (this.#ws.readyState === WebSocket.OPEN) {
      this.#ws.send(message);
    } else {
      this.#queue.push(message);
    }
  }
  
  on(type, handler) {
    if (!this.#handlers.has(type)) this.#handlers.set(type, new Set());
    this.#handlers.get(type).add(handler);
    return () => this.#handlers.get(type).delete(handler);
  }
  
  #startHeartbeat() {
    this.#heartbeatInterval = setInterval(() => {
      this.send('ping', {});
    }, 30000);
  }
  
  #stopHeartbeat() { clearInterval(this.#heartbeatInterval); }
  #flushQueue() { while (this.#queue.length) this.#ws.send(this.#queue.shift()); }
  
  #reconnect() {
    if (this.#reconnecting) return;
    this.#reconnecting = true;
    const delay = Math.min(1000 * Math.pow(2, Math.random() * 5), this.maxReconnectDelay);
    setTimeout(() => this.connect(), delay);
  }
}

// 代码示例 6：EventSource 与自动重连
function createEventSource(url, handlers) {
  const source = new EventSource(url);
  
  source.onopen = () => handlers.onConnect?.();
  source.onerror = () => handlers.onError?.();
  
  // 监听命名事件
  source.addEventListener('message', e => handlers.onMessage?.(JSON.parse(e.data)));
  source.addEventListener('notification', e => handlers.onNotification?.(JSON.parse(e.data)));
  
  return {
    close: () => source.close(),
    get readyState() { return source.readyState; }
  };
}

// 代码示例 7：竞态条件处理
function useLatest() {
  let currentId = 0;
  
  return async function fetchLatest(url) {
    const id = ++currentId;
    const response = await fetch(url);
    
    if (id !== currentId) {
      throw new Error('Stale request');
    }
    
    return response.json();
  };
}

// AbortController 组合
function createAbortGroup() {
  const controllers = new Set();
  
  return {
    create() {
      const controller = new AbortController();
      controllers.add(controller);
      controller.signal.addEventListener('abort', () => controllers.delete(controller));
      return controller;
    },
    abortAll() {
      controllers.forEach(c => c.abort());
      controllers.clear();
    }
  };
}

class APIClient {
  #baseUrl; #token; #abortGroup;
  
  constructor(baseUrl) {
    this.#baseUrl = baseUrl;
    this.#abortGroup = createAbortGroup();
  }
  
  setToken(token) { this.#token = token; }
  
  async get(path, params) {
    const url = new URL(path, this.#baseUrl);
    if (params) Object.entries(params).forEach(([k, v]) => url.searchParams.set(k, v));
    return this.#request(url, { method: 'GET' });
  }
  
  async post(path, body) {
    return this.#request(new URL(path, this.#baseUrl), {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(body)
    });
  }
  
  async #request(url, options) {
    const controller = this.#abortGroup.create();
    const response = await fetch(url, {
      ...options,
      signal: controller.signal,
      headers: { ...options.headers, ...(this.#token && { Authorization: `Bearer ${this.#token}` }) }
    });
    if (!response.ok) throw Object.assign(new Error(`HTTP ${response.status}`), { status: response.status });
    return response.json();
  }
  
  cancelAll() { this.#abortGroup.abortAll(); }
}

async function chunkedUpload(file, url, chunkSize = 5 * 1024 * 1024) {
  const totalChunks = Math.ceil(file.size / chunkSize);
  const uploadId = crypto.randomUUID();
  
  for (let i = 0; i < totalChunks; i++) {
    const start = i * chunkSize;
    const end = Math.min(start + chunkSize, file.size);
    const chunk = file.slice(start, end);
    
    await fetch(`${url}/upload`, {
      method: 'PUT',
      headers: {
        'X-Upload-Id': uploadId,
        'X-Chunk-Index': i,
        'X-Total-Chunks': totalChunks,
        'Content-Range': `bytes ${start}-${end - 1}/${file.size}`
      },
      body: chunk
    });
  }
  
  // 完成上传
  await fetch(`${url}/complete`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ uploadId, filename: file.name })
  });
}

async function longPoll(url, handler, options = {}) {
  const { signal, retryDelay = 1000 } = options;
  
  while (!signal?.aborted) {
    try {
      const response = await fetch(url, { signal });
      if (response.status === 204) continue; // 无新数据
      const data = await response.json();
      handler(data);
    } catch (e) {
      if (e.name === 'AbortError') break;
      await new Promise(r => setTimeout(r, retryDelay));
    }
  }
}

通信方式	方向	协议	适用场景
Fetch	客户端→服务器	HTTP	REST API
WebSocket	双向	WS	实时通信
SSE	服务器→客户端	HTTP	推送通知
WebRTC	点对点	UDP/SCTP	音视频/数据
Beacon	客户端→服务器	HTTP	页面卸载时发送