快捷導(dǎo)航

JavaScript中fetch流式實現(xiàn)中斷重連

更新時間：2026年03月12日 10:27:04 作者：豈不聞

在JavaScript中,當(dāng)使用fetch API發(fā)送網(wǎng)絡(luò)請求時可能會遇到中斷的錯誤,這篇文章主要介紹了JavaScript中fetch流式實現(xiàn)中斷重連的相關(guān)資料,文中通過代碼介紹的非常詳細,需要的朋友可以參考下

SSE(流式數(shù)據(jù))不用原生EventSource而是用fetch

需要認證，靈活控制，或非標準模式時候使用fetch

對比項	EventSource	fetch + ReadableStream
? 自動重連	是（有限）	否（需手動）
? 自動解析 `data:`/`id:`	是	否（需手動解析）
? 自定義 Header	否（無法加 `Authorization` 等）	? 可以
? 無法中止連接（舊版）	需調(diào)用 `.close()`	? 可用 `AbortController` 精確控制
? 無法處理非標準格式	僅支持 `text/event-stream`	? 可處理任意流（JSONL、自定義文本等）
? 無法獲取響應(yīng)頭	否	? 可讀取 `response.headers`

流式響應(yīng)

服務(wù)端立馬不關(guān)閉；

持續(xù)通過Connection:keep-alive保持TCP連接；

分塊發(fā)送數(shù)據(jù)(Transfer-Encoding:chunked)；

客戶端通過response.body.getReader()分塊讀取；

const response = await fetch('/stream');
const reader = response.body.getReader();

while (true) {
  const { done, value } = await reader.read();
  if (done) break; // 連接關(guān)閉
  console.log('收到數(shù)據(jù)塊:', value); // Uint8Array
}

一旦reader.read()非正常中斷或報錯，重新發(fā)起fetch請求，HTTP是無狀態(tài)的，不能恢復(fù)一個已經(jīng)斷開的TCP連接。

中斷：網(wǎng)絡(luò)丟失(網(wǎng)絡(luò)切換或丟失)，服務(wù)端重啟或者崩潰或其他的原因?qū)е铝魇綌?shù)據(jù)未完成而中斷；

重連：中斷后重新發(fā)起全新的請求；

重連的兩種機制（斷點續(xù)傳，指數(shù)退避）

斷點續(xù)傳-Last-Event-ID

SSE協(xié)議的標準做法，使用fetch也可以使用這種進行重連。

前端：記錄每條消息也就是每條消息塊的ID，重連時通過ID進行請求頭發(fā)送。

GET /events HTTP/1.1
Last-Event-ID: abc123

后端：讀取Last-Event-ID頭，從該ID后的消息開始推送，避免重復(fù)或丟失消息。

指數(shù)退避-Exponential Backoff

在中斷發(fā)起重連時，通過時間間隔進行重連比如 (第一次/1s 第二次/2s 第三次/4s 第四次/8s)成指數(shù)的形式時間間隔避免時間間隔太短帶給服務(wù)器壓力，設(shè)置最大時間不超過30s

delay = min( base * 2^retryCount + random_jitter, max_delay )

代碼例子

前端fetchServe.js

// src/utils/fetchServe.js

/**
 * 增強版 SSE 客戶端（基于 fetch）
 * 特性：
 * - 自動檢測中斷并重連（指數(shù)退避）
 * - 支持 Last-Event-ID 斷點續(xù)傳
 * - 提供 onStatus 回調(diào)（connecting / connected / disconnected）
 * - 完全兼容 Vue 2 生命周期
 */
export class FetchSSEClient {
  constructor(url, { onMessage, onError, onStatus }) {
    this.url = url;
    this.onMessage = onMessage || (() => {});
    this.onError = onError || console.error;
    this.onStatus = onStatus || (() => {});

    this.lastEventId = null;
    this.abortController = null;
    this.reconnectTimer = null;
    this.retryCount = 0;
    this.isClosing = false;
    this.isConnected = false;
  }

  start() {
    if (this.isClosing) return;
    this.connect();
  }

  async connect() {
    this.cleanup();

    // 更新狀態(tài)：正在連接
    this.updateStatus('connecting');

    const controller = new AbortController();
    this.abortController = controller;

    try {
      const headers = {};
      if (this.lastEventId) {
        headers['Last-Event-ID'] = this.lastEventId;
      }

      const response = await fetch(this.url, {
        method: 'GET',
        headers: {
          Accept: 'text/event-stream',
          Connection: 'keep-alive',
          ...headers,
        },
        signal: controller.signal,
      });

      if (!response.ok || !response.body) {
        throw new Error(`HTTP ${response.status}`);
      }

      const reader = response.body.getReader();
      const decoder = new TextDecoder();
      let buffer = '';

      this.retryCount = 0;
      this.isConnected = true;
      this.updateStatus('connected'); // 連接成功！

      while (true) {
        const { done, value } = await reader.read();

        if (done) {
          // 服務(wù)端主動關(guān)閉（如 Nginx 超時、后端重啟）
          console.warn('[SSE] 服務(wù)端關(guān)閉了連接');
          this.handleDisconnect();
          break;
        }

        if (value?.length) {
          buffer += decoder.decode(value, { stream: true });
          const lines = buffer.split('\n');
          buffer = lines.pop() || '';

          let currentId = '';
          let currentData = '';

          for (const line of lines) {
            const trimmed = line.trim();
            if (trimmed.startsWith('id:')) {
              currentId = trimmed.slice(3).trim();
            } else if (trimmed.startsWith('data:')) {
              currentData = trimmed.slice(5).trim();
            } else if (trimmed === '') {
              if (currentData) {
                try {
                  const parsed = JSON.parse(currentData);
                  if (currentId) this.lastEventId = currentId;
                  this.onMessage(parsed, currentId);
                } catch (e) {
                  console.warn('SSE 解析失敗:', currentData);
                }
                currentData = '';
                currentId = '';
              }
            }
          }
        }
      }
    } catch (err) {
      if (controller.signal.aborted) return; // 正常關(guān)閉

      console.error('[SSE] 連接異常:', err.message);
      this.onError(err);
      this.handleDisconnect();
    }
  }

  /**
   * 統(tǒng)一處理連接斷開邏輯
   */
  handleDisconnect() {
    this.isConnected = false;
    this.updateStatus('disconnected');
    this.scheduleReconnect();
  }

  /**
   * 調(diào)度自動重連（指數(shù)退避 + 隨機抖動）
   */
  scheduleReconnect() {
    if (this.isClosing || this.reconnectTimer) return;

    const baseDelay = 1000;
    const maxDelay = 30000;
    const delay = Math.min(
      baseDelay * Math.pow(2, this.retryCount) + Math.random() * 1000,
      maxDelay
    );

    this.reconnectTimer = setTimeout(() => {
      console.log(`[SSE] 自動重連中... 第 ${this.retryCount + 1} 次嘗試`);
      this.reconnectTimer = null;
      this.retryCount++;
      this.start();
    }, delay);
  }

  updateStatus(status) {
    this.onStatus(status); // 通知外部（如 Vue 組件更新 UI）
  }

  close() {
    this.isClosing = true;
    this.cleanup();
    this.updateStatus('closed');
  }

  cleanup() {
    if (this.abortController) {
      this.abortController.abort();
      this.abortController = null;
    }
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer);
      this.reconnectTimer = null;
    }
    this.retryCount = 0;
  }
}

vue2組件EventSource.vue

<!-- src/components/AutoReconnectSSE.vue -->
<template>
  <div style="padding: 20px; font-family: Arial, sans-serif;">
    <h2>自動重連 SSE 示例</h2>
    
    <!-- 狀態(tài)指示器 -->
    <div :style="{ color: statusColor }">
      ● 狀態(tài): {{ statusText }}
    </div>

    <!-- 最新消息 -->
    <div v-if="latestMessage" style="margin-top: 15px; padding: 10px; background: #f9f9f9;">
      <strong>最新消息 (ID: {{ lastId }}):</strong>
      <pre>{{ JSON.stringify(latestMessage, null, 2) }}</pre>
    </div>

    <!-- 手動重連按鈕（僅用于調(diào)試） -->
    <button 
      v-if="status !== 'connected'" 
      @click="forceReconnect"
      style="margin-top: 10px; padding: 6px 12px;"
    >
      強制重連
    </button>
  </div>
</template>

<script>
import { FetchSSEClient } from '@/utils/FetchSSEClient';

// 狀態(tài)映射
const STATUS_MAP = {
  connecting: { text: '連接中...', color: '#ffa500' },
  connected: { text: '已連接', color: '#00aa00' },
  disconnected: { text: '連接中斷，正在重連...', color: '#ff6600' },
  closed: { text: '已關(guān)閉', color: '#888' },
};

export default {
  name: 'AutoReconnectSSE',
  data() {
    return {
      status: 'connecting',
      latestMessage: null,
      lastId: null,
      sseClient: null,
    };
  },

  computed: {
    statusText() {
      return STATUS_MAP[this.status]?.text || '未知';
    },
    statusColor() {
      return STATUS_MAP[this.status]?.color || '#000';
    }
  },

  created() {
    this.initSSE();
  },

  beforeDestroy() {
    // 清理 防止內(nèi)存泄漏和重復(fù)連接
    if (this.sseClient) {
      this.sseClient.close();
    }
  },

  methods: {
    initSSE() {
      this.sseClient = new FetchSSEClient('http://localhost:3000/api/events', {
        onMessage: (data, id) => {
          this.latestMessage = data;
          this.lastId = id;
        },
        onError: (error) => {
          console.error('SSE 錯誤:', error);
        },
        onStatus: (status) => {
          // 自動同步狀態(tài)到 Vue 響應(yīng)式數(shù)據(jù)
          this.status = status;
        }
      });

      this.sseClient.start(); // 啟動自動連接
    },

    forceReconnect() {
      if (this.sseClient) {
        this.sseClient.close();
      }
      this.status = 'connecting';
      this.initSSE();
    }
  }
};
</script>

node.js模擬后端服務(wù)

// server.js
const express = require('express');
const cors = require('cors');

const app = express();

// 啟用 CORS，允許前端跨域請求，并支持自定義 Header（如 Last-Event-ID）
app.use(cors({
  origin: 'http://localhost:8080', // Vue DevServer 默認地址
  credentials: true,
  // 暴露 Last-Event-ID 給前端（雖然本例中前端不讀響應(yīng)頭，但良好實踐）
  exposedHeaders: ['Last-Event-ID'],
}));

// 模擬一個全局消息隊列（生產(chǎn)環(huán)境應(yīng)替換為 Redis、Kafka 或數(shù)據(jù)庫）
let globalMessageId = 0;
const messageHistory = []; // 保存最近的消息，用于斷點續(xù)傳

/**
 * 生成一條新消息并存入歷史記錄
 * @returns {{id: string, data: object}} 新消息對象
 */
function generateMessage() {
  globalMessageId++;
  const msg = {
    id: String(globalMessageId), // SSE 要求 id 是字符串
    data: {
      timestamp: new Date().toISOString(),
      value: Math.floor(Math.random() * 100),
      messageId: globalMessageId,
    }
  };
  messageHistory.push(msg);
  // 限制內(nèi)存占用：只保留最近 100 條
  if (messageHistory.length > 100) messageHistory.shift();
  return msg;
}

// 初始化幾條消息，確?？蛻舳耸状芜B接就有數(shù)據(jù)
for (let i = 0; i < 5; i++) generateMessage();

/**
 * SSE 流式接口：/api/events
 * 支持標準 SSE 格式 + Last-Event-ID 斷點續(xù)傳
 */
app.get('/api/events', (req, res) => {
  // 從請求頭中讀取客戶端上次收到的最后消息 ID
  const lastId = req.headers['last-event-id'] || null;
  console.log(`[SSE] 客戶端重連，Last-Event-ID: ${lastId}`);

  // 設(shè)置響應(yīng)頭 —— 這是 SSE 的核心！
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',     // 告訴瀏覽器這是 SSE 流
    'Cache-Control': 'no-cache',             // 禁用緩存
    'Connection': 'keep-alive',              // 保持 TCP 連接
    'Access-Control-Allow-Origin': '*',      // 允許跨域（生產(chǎn)環(huán)境應(yīng)指定域名）
    'X-Accel-Buffering': 'no',               // 關(guān)鍵！禁用 Nginx/Apache 緩沖
  });

  // 確定從哪條消息開始推送（實現(xiàn)斷點續(xù)傳）
  let startIndex = 0;
  if (lastId) {
    // 在歷史消息中查找 lastId 的位置
    const lastIdx = messageHistory.findIndex(m => m.id === lastId);
    if (lastIdx !== -1) {
      startIndex = lastIdx + 1; // 從下一條開始發(fā)送，避免重復(fù)
    }
    // 如果 lastId 太舊（不在 history 中），則從最新或開頭發(fā)（按業(yè)務(wù)需求調(diào)整）
  }

  // 推送缺失的歷史消息（確保不丟消息）
  for (let i = startIndex; i < messageHistory.length; i++) {
    const msg = messageHistory[i];
    // SSE 標準格式：每條消息以 "id:" 和 "data:" 開頭，空行結(jié)束
    res.write(`id: ${msg.id}\n`);
    res.write(`data: ${JSON.stringify(msg.data)}\n\n`);
  }

  // 啟動定時器，持續(xù)生成并推送新消息
  const interval = setInterval(() => {
    const msg = generateMessage();
    res.write(`id: ${msg.id}\n`);
    res.write(`data: ${JSON.stringify(msg.data)}\n\n`);
    // 注意：res.write() 不會自動 flush，但 Node.js 通常會及時發(fā)送
  }, 2000); // 每 2 秒推送一次

  // 監(jiān)聽客戶端斷開連接（如關(guān)閉標簽頁、網(wǎng)絡(luò)中斷）
  req.on('close', () => {
    console.log('[SSE] 客戶端斷開連接，清理定時器');
    clearInterval(interval); // 停止生成新消息
    res.end();              // 顯式關(guān)閉響應(yīng)
  });
});

// 啟動服務(wù)器
const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
  console.log(`SSE 服務(wù)已啟動，請訪問 http://localhost:${PORT}`);
});

根據(jù)實際對接接口以及需求來參考進行開發(fā)