在Angular中展示Spring Boot WebFlux的APPLICATION_STREAM_JSON对象报错

这个问题我之前帮朋友排查过类似的，核心是Angular 4的HttpClient默认不擅长处理Spring Boot WebFlux输出的application/stream+json这种流式JSON响应——毕竟前者默认是等着拿完整的JSON包，而后者是每隔几秒就吐一个单独的JSON对象，用换行分隔，这就导致HttpClient在拿到部分数据时就尝试解析成完整JSON，自然会报错。

先确认你的Spring Boot后端接口是正确的，比如类似这样：

@GetMapping(value = "/data-stream", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Flux<YourDataClass> generateStream() {
    // 每5秒生成一个数据对象
    return Flux.interval(Duration.ofSeconds(5))
            .map(tick -> new YourDataClass("Sample Data " + tick));
}

如果这段没问题，那接下来看Angular端的两种修复方案：

方案一：直接处理application/stream+json流式响应 #

我们需要告诉HttpClient不要自动解析JSON，而是把响应当成文本流处理，然后自己分割每行的JSON对象并解析：

1. 在Angular组件中编写请求逻辑：

import { HttpClient, HttpResponse } from '@angular/common/http';
import { Observable } from 'rxjs/Observable';
import { map, filter, scan } from 'rxjs/operators';

// 假设你的数据类型定义
interface YourDataClass {
  content: string;
}

@Component({
  selector: 'app-stream-view',
  templateUrl: './stream-view.component.html'
})
export class StreamViewComponent {
  dataStream$: Observable<YourDataClass[]>;

  constructor(private http: HttpClient) {
    this.dataStream$ = this.http.get('/api/data-stream', { 
      responseType: 'text', 
      observe: 'events' 
    }).pipe(
      // 只处理数据块事件
      filter(event => event.type === 'data'),
      // 提取文本内容
      map(event => (event as HttpResponse<string>).body),
      // 累积分块内容，避免单个JSON被拆分到多个块里
      scan((accumulatedText, newChunk) => accumulatedText + newChunk, ''),
      // 按换行分割成单独的JSON行，过滤空行
      map(fullText => fullText.split('\n').filter(line => line.trim() !== '')),
      // 把每行解析成JSON对象
      map(jsonLines => jsonLines.map(line => JSON.parse(line) as YourDataClass))
    );
  }
}

2. 在模板中用async管道展示数据：

<h3>流式数据展示</h3>
<div *ngFor="let data of dataStream$ | async">
  {{ data.content }}
</div>

方案二：切换为SSE（Server-Sent Events）协议 #

如果可以调整后端接口，换成SSE协议会更适配浏览器的流式处理，Angular端的实现也更简洁：

1. 修改Spring Boot后端接口为SSE格式：

import org.springframework.http.codec.ServerSentEvent;
import reactor.core.publisher.Flux;
import java.time.Duration;

@GetMapping(value = "/sse-data-stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<YourDataClass>> generateSseStream() {
    return Flux.interval(Duration.ofSeconds(5))
            .map(tick -> ServerSentEvent.<YourDataClass>builder()
                    .id(String.valueOf(tick))
                    .event("new-data")
                    .data(new YourDataClass("SSE Data " + tick))
                    .build());
}

2. Angular端用EventSource处理SSE：

import { Observable } from 'rxjs/Observable';

interface YourDataClass {
  content: string;
}

@Component({
  selector: 'app-sse-view',
  templateUrl: './sse-view.component.html'
})
export class SseViewComponent {
  dataStream$: Observable<YourDataClass>;

  constructor() {
    const eventSource = new EventSource('/api/sse-data-stream');
    
    this.dataStream$ = Observable.create(observer => {
      // 监听自定义的new-data事件
      eventSource.addEventListener('new-data', (event: MessageEvent) => {
        const data = JSON.parse(event.data) as YourDataClass;
        observer.next(data);
      });

      // 处理错误或连接关闭
      eventSource.addEventListener('error', (error) => {
        if (eventSource.readyState === EventSource.CLOSED) {
          console.log('SSE连接已关闭');
        } else {
          observer.error(error);
        }
      });

      // 组件销毁时关闭连接
      return () => {
        eventSource.close();
      };
    });
  }
}

3. 模板展示同样用async管道：

<h3>SSE流式数据展示</h3>
<div *ngFor="let data of dataStream$ | async">
  {{ data.content }}
</div>

为什么application/json能正常工作？ #

当你切换为常规application/json时，WebFlux会把所有数据攒成一个JSON数组一次性返回，HttpClient可以正常解析这个完整的JSON响应，自然不会报错——但这就失去了流式传输的意义啦。

内容的提问来源于stack exchange，提问作者Jan Testowy