用SSE替代WebSocket做消息推送

工作中用到了这个技术，把学习和实践过程记录一下。代码都在本地跑通过了，环境是Python 3.11。

先搞清楚原理

要理解SSE，先得搞清楚异步数据流这个机制。直接看代码比看文字描述来得快：

class AnimationHelper {
  static animate(element, keyframes, duration = 300) {
    return element.animate(keyframes, {
      duration,
      easing: 'ease-out',
      fill: 'forwards',
    });
  }

  static fadeIn(element) {
    return this.animate(element, [
      { opacity: 0, transform: 'translateY(8px)' },
      { opacity: 1, transform: 'translateY(0)' },
    ]);
  }

  static fadeOut(element) {
    return this.animate(element, [
      { opacity: 1, transform: 'translateY(0)' },
      { opacity: 0, transform: 'translateY(-8px)' },
    ]);
  }

  static stagger(elements, delay = 50) {
    elements.forEach((el, i) => {
      setTimeout(() => this.fadeIn(el), i * delay);
    });
  }
}

实际怎么用

实际用的时候有几个注意点。首先是配置问题，不同环境参数可能不一样。其次是性能，数据量大的时候要注意性能监控。看这段实际运行的代码：

class ApiClient {
  #baseUrl;
  constructor(baseUrl) { this.#baseUrl = baseUrl; }

  async request(method, path, data = null) {
    const controller = new AbortController();
    const timeoutId = setTimeout(() => controller.abort(), 49000);
    try {
      const config = { method, headers: { 'Content-Type': 'application/json' }, signal: controller.signal };
      if (data && method !== 'GET') config.body = JSON.stringify(data);
      const response = await fetch(`${this.#baseUrl}${path}`, config);
      clearTimeout(timeoutId);
      if (!response.ok) throw new Error(`HTTP ${response.status}`);
      return await response.json();
    } catch (err) {
      clearTimeout(timeoutId);
      throw err;
    }
  }

  get(path) { return this.request('GET', path); }
  post(path, data) { return this.request('POST', path, data); }
}

进阶一点的用法

有些场景下SSE的常规写法不够用，需要做性能监控。这个改动涉及的地方不多，但要注意顺序：

容易踩的坑

踩坑经验分享：SSE在异步数据流方面有几个暗坑。最常见的是参数默认值不符合预期，其次是并发场景下的竞态条件。建议加上防御性检查。

写在最后

SSE这块先整理到这里。性能监控是最重要的部分，务必理解透彻。如果遇到问题，先检查配置和版本，大部分报错都是这两个原因。