Angular 16中SSE流式数据无法实时展示的技术求助

问题背景 #

基于SSE（Content-Type: text/event-stream）开发流式API，使用Angular 16框架。当前代码可接收流式数据，但仅在所有数据接收完成后才展示最后一批内容，无法实现数据一到达就立刻渲染。尝试过ngZone、变更检测策略等方法均无效，需可行技术方案。

现有代码 #

stream.service.ts #

connectToSse(): Observable<any> {
    return new Observable((observer) => {
      this.eventSource = new EventSource(`${this.url}/getStreamingData`);
      this.eventSource.onmessage = (event) => {
        // 处理字符串数据转换为 [object, object] 的逻辑
          observer.next(ArrayObj)
      };

      this.eventSource.onerror = (error) => {
        observer.error(error);
        this.closeConnection();
      };
    });
  }

html.component.ts #

getStreamData() {
    this.sseSubscription = this._stream.connectToSse().subscribe(
      (event: any) => {
            // 能逐个（或批量）接收解析后的数组对象
          this.accumulatedData$ = of(event);
      },
      (error: any) => {
        console.error('SSE error:', error);
      }
    );
  }

HTML模板 #

<div class="json-container">
  <div *ngFor="let row of accumulatedData$ | async;">    
    <ng-container *ngFor="let item of row | keyvalue">
      <div class="">
        {{item.key}} || {{item.value}}
      </div>
    </ng-container>
    <hr />
  </div>
</div>

问题根源 #

核心问题是组件每次收到新数据时，都用of(event)创建新Observable并赋值给accumulatedData$，而非在同一个流中累加数据。Angular的async管道会订阅新Observable，旧订阅被取消导致之前数据被覆盖，且仅当最后一次Observable完成（或SSE连接关闭）时才触发渲染。

解决方案 #

1. 用Subject管理持续数据流 #

在组件中使用BehaviorSubject累积数据，替代每次创建新Observable：

import { BehaviorSubject, Subscription } from 'rxjs';

// 组件类内定义
private accumulatedDataSubject = new BehaviorSubject<any[]>([]);
accumulatedData$ = this.accumulatedDataSubject.asObservable();
sseSubscription?: Subscription;

getStreamData() {
    this.sseSubscription = this._stream.connectToSse().subscribe(
      (event: any) => {
            // 追加新数据到现有数组
            const currentData = this.accumulatedDataSubject.value;
            const newData = Array.isArray(event) 
              ? [...currentData, ...event] 
              : [...currentData, event];
            this.accumulatedDataSubject.next(newData);
      },
      (error: any) => {
        console.error('SSE error:', error);
        this.accumulatedDataSubject.complete();
      },
      () => {
        // SSE连接关闭时完成Subject
        this.accumulatedDataSubject.complete();
      }
    );
  }

// 组件销毁时清理资源
ngOnDestroy() {
  this.sseSubscription?.unsubscribe();
  this.accumulatedDataSubject.complete();
}

2. 调整HTML模板适配累加数组 #

模板直接遍历累积后的数组即可：

<div class="json-container">
  <div *ngFor="let row of accumulatedData$ | async;">    
    <ng-container *ngFor="let item of row | keyvalue">
      <div class="">
        {{item.key}} || {{item.value}}
      </div>
    </ng-container>
    <hr />
  </div>
</div>

3. 优化SSE服务的变更检测（可选） #

若仍有渲染延迟，在服务中用NgZone包裹推送逻辑，确保Angular捕获变更：

import { Injectable, NgZone } from '@angular/core';

constructor(private ngZone: NgZone) {}

connectToSse(): Observable<any> {
    return new Observable((observer) => {
      this.eventSource = new EventSource(`${this.url}/getStreamingData`);
      this.eventSource.onmessage = (event) => {
        // 数据转换逻辑
        this.ngZone.run(() => {
          observer.next(ArrayObj);
        });
      };

      this.eventSource.onerror = (error) => {
        this.ngZone.run(() => {
          observer.error(error);
          this.closeConnection();
        });
      };
    });
  }

验证步骤 #

1. 启动后端SSE服务，确保每次推送小批量或单个数据对象
2. 启动Angular应用并调用getStreamData()
3. 观察页面，每收到一批数据会立刻渲染新增内容，无需等待全部流完成

内容的提问来源于stack exchange，提问作者Deepak-ranolia