通常由三种方案

方案	Spring MVC	Spring WebFlux	适用场景
StreamingResponseBody	✅	❌	下载大文件、持续输出文本
SseEmitter (SSE)	✅	❌	ChatGPT 类似的单向推送
Flux<T> + ServerSentEvent	❌	✅	Reactive 流式推送，推荐新项目

Spring MVC: StreamingResponseBody

Server Controller

    @GetMapping(value="/stream", produces=MediaType.TEXT_PLAIN_VALUE)
    public StreamingResponseBody stream() {
        return outputStream -> {
            for (int i = 0; i<10; i++) {
                String msg = "message-" + i + "\n";
                outputStream.write(msg.getBytes());
                outputStream.flush();
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
    }

React Client

import { Button, Typography } from "antd";
import { useState } from "react";

const { Text } = Typography;

// Code is complex, however we can wrap it to a util function.
async function* streamLines(response: Response) {
  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() || '';

    yield* lines;
  }
}

export const StreamExample: React.FC = () => {
  const [messages, setMessages] = useState<string[]>([]);

  const handleClick = async () => {
    const response = await fetch('/lhn/api/hello/stream');
    for await (const line of streamLines(response)) {
      setMessages(prev => [...prev, line]);
    }
  }
  
  return (
    <div>
      <Button onClick={handleClick}>
        Stream
      </Button>
      <div className='flex flex-col gap-2'>
        {messages.map((message, index) => (
          <Text key={index}>{message}</Text>
        ))}
      </div>
    </div>
  );
};

Spring MVC: SseEmitter

Server Controller

返回值是 SseEmitter，没有类型。

//    @GetMapping(value="/sse-emitter", produces=MediaType.TEXT_EVENT_STREAM_VALUE)
    @PostMapping(value="/sse-emitter", produces=MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter sseEmitter(@RequestBody String prompt) {
        log.info("Prompt: " + prompt);
        
        SseEmitter emitter = new SseEmitter(0L);
        Executors.newSingleThreadExecutor().submit(() -> {
            try {
//                emitter.send("Hello");
//                Thread.sleep(1000);
//                emitter.send("World");
//                Thread.sleep(1000);
//                emitter.send("!");
                
                emitter.send(SseEmitter.event().name("message").data("Hello"));
                Thread.sleep(1000);
                emitter.send(SseEmitter.event().name("message").data("Wrold"));
                Thread.sleep(1000);
                emitter.send(SseEmitter.event().name("message").data("!"));
                
                emitter.complete();
            } catch (Exception e) {
                emitter.completeWithError(e);
            }
        });
        return emitter;
    }

React Client

开发环境中使用 UmiJs proxy，它会 cache respnse，导致不会一有 message 就更新。

可以使用 @microsoft/fetch-event-source 简化前端的解析。

import { useState, useRef, useEffect, FC } from "react";
import { Button, Typography } from "antd";
import { getErrorMsg } from "@/utils/errors";
import { fetchEventSource } from "@microsoft/fetch-event-source";

const { Text } = Typography;

/** 
 * A test component for SSE (Server-Sent Events) emitter.
 * Two ways of message format: direct message or event.
 * Both client and server can close the connection.
 * 
 * UmiJs local proxy has cache. It compresses and caches multiple responses.
 * 
 * Use @microsoft/fetch-event-source to support POST prompt and simplify client code.
 * Use AbortSignal to cancel the request in client side.
 */
export const SseEmitterTest: FC = () => {
  const [messages, setMessages] = useState<string[]>([]);
  const abortRef = useRef<AbortController | null>(null);

  useEffect(() => {
    return () => {
      abortRef.current?.abort();
    };
  }, []);

  const handleClick = () => {
    setMessages([]);
    
    const es = new EventSource('/lhn/api/hello/sse-emitter');
    es.onmessage = (event) => {
      console.log('Received message:', new Date().toLocaleTimeString(), event.data);
      setMessages(prev => [...prev, event.data]);
    };
    es.onerror = (error) => {
      setMessages(prev => [...prev, getErrorMsg(error) || 'EventSource failed']);
      es.close();
    };
  };

  const handleClick2 = async (abortSignal?: AbortSignal) => {
    setMessages([]);
    abortRef.current?.abort();
    const ctrl = new AbortController();
    abortRef.current = ctrl;

    await fetchEventSource('/lhn/api/hello/sse-emitter', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({
        prompt: 'Hello'
      }),
      signal: abortSignal ?? ctrl.signal,

      onmessage(event) {
        setMessages(prev => [...prev, event.data]);
      }
    });
  }

  return (
    <div>
      <Button onClick={() => handleClick2()}>
        SSE Stream
      </Button>
      <div className='flex flex-col gap-2'>
        {messages.map((message, index) => (
          <Text key={index}>{message}</Text>
        ))}
      </div>
    </div>
  );
};

Spring WebFlux

Server Controller

Response 有类型，写法简单。

    @Data
    private static final class ChatRequest {
        private String prompt;
    }
    
    @Data
    @AllArgsConstructor
    private static final class ChatResponse {
        private String content;
    }
    
    @PostMapping(value="flux-stream", produces=MediaType.APPLICATION_NDJSON_VALUE)
    public Flux<ChatResponse> fluxMessage(@RequestBody ChatRequest request) {
        log.info("Request: " + request);
        return Flux.interval(Duration.ofSeconds(1))
                .take(5)
                .map(i -> new ChatResponse("Message: " + i + ", " + request.prompt));
    }

React Client

server 返回的 ndjson，前端不能使用 @microsoft/fetch-event-source（它只能接收 text/event-stream），自己写个 parseResponse 的 util。

import { Button, Typography } from "antd";
import { FC, useRef, useState } from "react";

const { Text } = Typography;

type ChatResponse = {
  content: string;
}

async function* parseResponse<T>(response: Response): AsyncGenerator<T> {
  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) as T;
      }
    }
  }
}


/**
 * Flex stream is same as stream. Server code is simple. Client usage is same as stream.
 */
export const FluxStream: FC = () => {
  const [messages, setMessages] = useState<string[]>([]);
  const abortControlRef = useRef<AbortController | null>(null);

  const handleClick = async () => {
    setMessages([]);
    abortControlRef.current = new AbortController();

    try {
      const response = await fetch('/lhn/api/hello/flux-stream', {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
        },
        body: JSON.stringify({
          prompt: 'Hello'
        }),
        signal: abortControlRef.current.signal,
      });

      for await (const resp of parseResponse<ChatResponse>(response)) {
        setMessages(prev => [...prev, resp.content]);
      }
    } catch (error) {
      if (error instanceof DOMException && error.name === 'AbortError') {
        return;
      }
      throw error;
    }
  }

  const handleStop = () => {
    abortControlRef.current?.abort();
  }

  return (
    <div className="flex gap-2">
      <Button onClick={handleClick}>
        Flux Stream
      </Button>
      <Button onClick={handleStop}>
        Stop
      </Button>
      <div className='flex flex-col gap-2'>
        {messages.map((message, index) => (
          <Text key={index}>{message}</Text>
        ))}
      </div>
    </div>
  );
};