Java 响应式编程：Flux 与 SseEmitter 深度解析

核心优势：

• ✅ 用户立即看到响应，体验提升 300%
• ✅ 线程快速释放，服务器吞吐量提升 5-10 倍
• ✅ 支持超长响应，不受 HTTP 超时限制
• ✅ 实时反馈进度，降低用户焦虑

二、技术背景与概念

2.1 SSE (Server-Sent Events)

官方定义： SSE 是 HTML5 标准的一部分，允许服务器主动向客户端推送数据。

核心特性：

• 基于 HTTP 协议，无需额外协议支持
• 单向通信（服务器→客户端）
• 自动重连机制
• 支持事件 ID 和自定义事件类型
• 纯文本协议，简单易用

数据格式：

data: 这是第一条消息

data: 这是第二条消息

event: custom-event
data: {"message": "JSON 数据"}
id: 123

2.2 Reactive Streams 与 Flux

Reactive Streams 是 JVM 上的响应式编程规范，定义了 4 个核心接口：

public interface Publisher<T> {
    void subscribe(Subscriber<? super T> subscriber);
}

public interface Subscriber<T> {
    void onSubscribe(Subscription subscription);
    void onNext(T item);
    void onError(Throwable throwable);
    void onComplete();
}

public interface Subscription {
    void request(long n); // 背压控制
    void cancel();
}

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {}

Flux 是 Project Reactor 对 Publisher 的实现，代表 0-N 个元素的异步序列：

Flux<T>: 0..N 个元素的流
├─ onNext(T) * N : 发射 N 个元素
├─ onError(Throwable): 发生错误（终止）
└─ onComplete() : 完成信号（终止）

2.3 两者的关系

SseEmitter (Spring MVC 实现)
├─ 基于 Servlet 异步支持
├─ 适合传统 Spring MVC 项目
└─ 不依赖响应式框架

Flux (响应式流)
├─ 基于 Reactive Streams 规范
├─ 需要 Spring WebFlux 支持
└─ 完整的响应式编程能力

三、SseEmitter 深度解析

3.1 基本使用

3.1.1 简单示例

@RestController
@RequestMapping("/api/v1/chat")
public class ChatController {
    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter streamMessage(@RequestParam String message) {
        // 创建 SseEmitter，设置超时时间为 5 分钟
        SseEmitter emitter = new SseEmitter(5 * 60 * 1000L);
        // 异步处理，避免阻塞主线程
        CompletableFuture.runAsync(() -> {
            try {
                // 模拟 AI 逐字生成回复
                String response = "这是一个流式响应示例";
                for (char c : response.toCharArray()) {
                    emitter.send(String.valueOf(c));
                    Thread.sleep(100); // 模拟生成延迟
                }
                emitter.complete(); // 发送完成信号
            } catch (Exception e) {
                emitter.completeWithError(e); // 发送错误信号
            }
        });
        return emitter;
    }
}

关键点说明：

1. produces = MediaType.TEXT_EVENT_STREAM_VALUE：必须设置 Content-Type 为 text/event-stream
2. SseEmitter(timeout)：超时时间，0 表示永不超时（不推荐）
3. CompletableFuture.runAsync()：异步执行，避免阻塞 Tomcat 线程
4. emitter.complete()：必须调用，否则客户端连接不会关闭

3.1.2 前端对接代码

// 原生 EventSource API
const eventSource = new EventSource('/api/v1/chat/stream?message=你好');
eventSource.onmessage = (event) => {
    console.log('收到数据:', event.data);
    document.getElementById('response').innerText += event.data;
};
eventSource.onerror = (error) => {
    console.error('连接错误:', error);
    eventSource.close();
};
// 监听完成事件（需要服务端发送特殊事件）
eventSource.addEventListener('complete', () => {
    console.log('流式响应完成');
    eventSource.close();
});

3.2 生产级实现

3.2.1 完整的聊天流式接口

@Slf4j
@RestController
@RequestMapping("/api/v1/chat")
@RequiredArgsConstructor
public class StreamChatController {
    private final ChatService chatService;
    private final ExecutorService executorService;

    @PostMapping(value = "/conversations/{sessionId}/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter streamChat(@PathVariable String sessionId, @RequestBody SendMessageReq req) {
        long startTime = System.currentTimeMillis();
        SseEmitter emitter = new SseEmitter(5 * 60 * 1000L); // 5 分钟超时

        // 设置回调处理
        emitter.onCompletion(() -> {
            log.info("SSE 连接正常完成，sessionId={}, cost={}ms", sessionId, System.currentTimeMillis() - startTime);
        });
        emitter.onTimeout(() -> {
            log.warn("SSE 连接超时，sessionId={}", sessionId);
            emitter.complete();
        });
        emitter.onError(throwable -> {
            log.error("SSE 连接异常，sessionId={}, error={}", sessionId, throwable.getMessage(), throwable);
        });

        // 异步处理消息
        executorService.execute(() -> {
            try {
                // 1. 发送开始事件
                emitter.send(SseEmitter.event().name("start").data(Map.of("messageId", generateMessageId())));
                // 2. 调用 AI 服务流式生成
                StringBuilder fullResponse = new StringBuilder();
                chatService.streamGenerate(sessionId, req.getContent(), chunk -> {
                    try {
                        fullResponse.append(chunk);
                        // 发送数据块
                        emitter.send(SseEmitter.event().name("message").data(Map.of("content", chunk)));
                    } catch (IOException e) {
                        throw new RuntimeException("发送数据失败", e);
                    }
                });
                // 3. 保存完整消息到数据库
                chatService.saveMessage(sessionId, fullResponse.toString());
                // 4. 发送完成事件
                emitter.send(SseEmitter.event().name("complete").data(Map.of(
                    "totalChunks", fullResponse.length(),
                    "costTime", System.currentTimeMillis() - startTime
                )));
                emitter.complete();
            } catch (Exception e) {
                log.error("流式生成失败", e);
                try {
                    emitter.send(SseEmitter.event().name("error").data(Map.of("message", e.getMessage())));
                } catch (IOException ioException) {
                    log.error("发送错误事件失败", ioException);
                }
                emitter.completeWithError(e);
            }
        });
        return emitter;
    }

    private String generateMessageId() {
        return "msg_" + System.currentTimeMillis();
    }
}

3.2.2 线程池配置

@Configuration
public class AsyncConfig {
    @Bean(name = "sseExecutor")
    public ExecutorService sseExecutor() {
        return new ThreadPoolExecutor(
            10, // 核心线程数
            50, // 最大线程数
            60L, TimeUnit.SECONDS, // 线程空闲时间
            new LinkedBlockingQueue<>(100), // 任务队列
            new ThreadFactoryBuilder().setNameFormat("sse-executor-%d").build(),
            new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
        );
    }
}

3.3 高级特性

3.3.1 自定义事件类型

// 服务端发送不同类型的事件
emitter.send(SseEmitter.event()
    .id("123") // 事件 ID（用于断线重连）
    .name("chat-message") // 自定义事件名
    .data(messageData) // 数据
    .comment("这是注释") // 注释（客户端不处理）
    .reconnectTime(3000L)); // 重连时间（毫秒）

// 客户端监听特定事件
eventSource.addEventListener('chat-message', (event) => {
    const data = JSON.parse(event.data);
    console.log('收到聊天消息:', data);
});

3.3.2 断线重连机制

@GetMapping(value = "/stream-with-resume", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter streamWithResume(@RequestHeader(value = "Last-Event-ID", required = false) String lastEventId) {
    SseEmitter emitter = new SseEmitter();
    // 从 lastEventId 位置继续发送
    int startIndex = lastEventId != null ? Integer.parseInt(lastEventId) : 0;
    CompletableFuture.runAsync(() -> {
        try {
            for (int i = startIndex; i < 100; i++) {
                emitter.send(SseEmitter.event()
                    .id(String.valueOf(i)) // 设置事件 ID
                    .data("数据块 " + i));
                Thread.sleep(100);
            }
            emitter.complete();
        } catch (Exception e) {
            emitter.completeWithError(e);
        }
    });
    return emitter;
}

// 客户端自动使用 Last-Event-ID 重连
const eventSource = new EventSource('/stream-with-resume');
// EventSource 会自动在重连时发送 Last-Event-ID 请求头

四、Flux 深度解析

4.1 引入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

注意： Flux 需要 Spring WebFlux 支持，可以与 Spring MVC 共存，但需要注意：

• WebFlux 运行在 Netty 上（默认），也可以配置为 Tomcat
• 两者可以同时存在，但推荐全面切换到 WebFlux 以发挥最大性能

4.2 基本使用

4.2.1 简单示例

@RestController
@RequestMapping("/api/v1/reactive")
public class ReactiveChatController {
    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> streamMessage(@RequestParam String message) {
        return Flux.interval(Duration.ofMillis(100)) // 每 100ms 发射一个元素
            .map(i -> "字符" + i)
            .take(10) // 只取前 10 个
            .doOnComplete(() -> log.info("流完成"));
    }
}

关键概念：

• Flux.interval()：周期性发射元素
• map()：转换元素
• take(n)：限制元素数量
• doOnComplete()：完成时的副作用操作

4.2.2 实际 AI 聊天场景

@RestController
@RequestMapping("/api/v1/reactive/chat")
@RequiredArgsConstructor
public class ReactiveStreamController {
    private final ReactiveAIService aiService;

    @PostMapping(value = "/conversations/{sessionId}/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<MessageChunk>> streamChat(@PathVariable String sessionId, @RequestBody SendMessageReq req) {
        return aiService.streamGenerate(req.getContent())
            .map(chunk -> ServerSentEvent.<MessageChunk>builder()
                .event("message")
                .data(new MessageChunk(chunk))
                .build())
            .concatWith(Flux.just(ServerSentEvent.<MessageChunk>builder()
                .event("complete")
                .data(new MessageChunk(""))
                .build()))
            .doOnSubscribe(sub -> log.info("客户端订阅，sessionId={}", sessionId))
            .doOnComplete(() -> log.info("流完成，sessionId={}", sessionId))
            .doOnError(e -> log.error("流异常，sessionId={}", sessionId, e));
    }
}

@Data
@AllArgsConstructor
class MessageChunk {
    private String content;
}

4.3 核心操作符详解

4.3.1 创建 Flux

// 1. 从集合创建
Flux<String> flux1 = Flux.fromIterable(Arrays.asList("A", "B", "C"));
// 2. 从数组创建
Flux<String> flux2 = Flux.fromArray(new String[]{"A", "B", "C"});
// 3. 从 Stream 创建
Flux<String> flux3 = Flux.fromStream(Stream.of("A", "B", "C"));
// 4. 动态生成
Flux<Integer> flux4 = Flux.generate(() -> 0,
    (state, sink) -> {
        sink.next(state); // 发射元素
        if (state == 10) sink.complete(); // 完成信号
        return state + 1; // 下一个状态
    });
// 5. 异步创建（最灵活）
Flux<String> flux5 = Flux.create(sink -> {
    // 模拟从外部回调接收数据
    externalService.onData(data -> sink.next(data));
    externalService.onComplete(() -> sink.complete());
    externalService.onError(error -> sink.error(error));
});
// 6. 定时器
Flux<Long> flux6 = Flux.interval(Duration.ofSeconds(1)); // 每秒发射一个递增的 Long

4.3.2 转换操作符

Flux<String> source = Flux.just("hello", "world");
// 1. map - 一对一转换
Flux<String> mapped = source.map(String::toUpperCase); // "hello" -> "HELLO", "world" -> "WORLD"
// 2. flatMap - 一对多转换（异步）
Flux<String> flatMapped = source.flatMap(word -> Flux.fromArray(word.split(""))); // "hello" -> ["h","e","l","l","o"]
// 3. concatMap - 一对多转换（保序）
Flux<String> concatMapped = source.concatMap(word -> Flux.fromArray(word.split("")));
// 4. filter - 过滤
Flux<String> filtered = source.filter(s -> s.length() > 4);
// 5. distinct - 去重
Flux<String> distinct = Flux.just("A", "B", "A").distinct();
// 6. take/skip - 限制
Flux<String> taken = source.take(5); // 取前 5 个
Flux<String> skipped = source.skip(2); // 跳过前 2 个
// 7. buffer - 批处理
Flux<List<String>> buffered = source.buffer(10); // 每 10 个元素打包成一个 List

4.3.3 组合操作符

Flux<String> flux1 = Flux.just("A", "B");
Flux<String> flux2 = Flux.just("C", "D");
// 1. concat - 顺序连接
Flux<String> concat = Flux.concat(flux1, flux2); // 输出：A, B, C, D
// 2. merge - 交错合并（异步）
Flux<String> merged = Flux.merge(flux1, flux2); // 输出：A, C, B, D (顺序不确定)
// 3. zip - 配对组合
Flux<String> zipped = Flux.zip(flux1, flux2, (a, b) -> a + b); // 输出："AC", "BD"
// 4. combineLatest - 最新值组合
Flux<String> combined = Flux.combineLatest(flux1, flux2, (a, b) -> a + b);

4.3.4 错误处理

Flux<String> flux = Flux.just("A", "B", "C").map(s -> {
    if (s.equals("B")) throw new RuntimeException("错误");
    return s;
});
// 1. onErrorReturn - 错误时返回默认值
Flux<String> handled1 = flux.onErrorReturn("默认值");
// 2. onErrorResume - 错误时切换到备用流
Flux<String> handled2 = flux.onErrorResume(e -> Flux.just("备用 1", "备用 2"));
// 3. onErrorContinue - 错误时跳过并继续
Flux<String> handled3 = flux.onErrorContinue((e, obj) -> log.error("处理 {} 时出错：{}", obj, e.getMessage()));
// 4. retry - 重试
Flux<String> retried = flux.retry(3); // 失败时重试 3 次
// 5. retryWhen - 自定义重试策略
Flux<String> retriedWhen = flux.retryWhen(Retry.backoff(3, Duration.ofSeconds(1))); // 指数退避重试

4.4 背压（Backpressure）处理

背压是响应式编程的核心特性，用于处理生产者速度 > 消费者速度的情况。

Flux<Integer> fastProducer = Flux.range(1, 1000).delayElements(Duration.ofMillis(1)); // 每毫秒生产一个
// 1. onBackpressureBuffer - 缓冲（可能 OOM）
Flux<Integer> buffered = fastProducer
    .onBackpressureBuffer(100, dropped -> log.warn("丢弃：{}", dropped));
// 2. onBackpressureDrop - 丢弃新元素
Flux<Integer> dropped = fastProducer
    .onBackpressureDrop(dropped -> log.warn("丢弃：{}", dropped));
// 3. onBackpressureLatest - 只保留最新元素
Flux<Integer> latest = fastProducer.onBackpressureLatest();
// 4. onBackpressureError - 抛出异常
Flux<Integer> error = fastProducer.onBackpressureError();

4.5 完整生产案例

@Service
@Slf4j
@RequiredArgsConstructor
public class ReactiveAIService {
    private final OpenAIClient openAIClient;
    private final MessageRepository messageRepository;

    /**
     * 流式生成 AI 回复
     */
    public Flux<String> streamGenerate(String sessionId, String prompt) {
        return Flux.create(sink -> {
            String messageId = generateMessageId();
            StringBuilder fullResponse = new StringBuilder();
            try {
                // 调用 OpenAI 流式 API
                openAIClient.streamChatCompletion(prompt, new StreamCallback() {
                    @Override
                    public void onChunk(String chunk) {
                        fullResponse.append(chunk);
                        sink.next(chunk); // 发射数据块
                    }

                    @Override
                    public void onComplete() {
                        // 保存完整消息
                        saveMessage(sessionId, messageId, fullResponse.toString()).subscribe(
                            saved -> log.info("消息保存成功：{}", messageId),
                            error -> log.error("消息保存失败", error)
                        );
                        sink.complete(); // 完成信号
                    }

                    @Override
                    public void onError(Throwable error) {
                        sink.error(error); // 错误信号
                    }
                });
            } catch (Exception e) {
                sink.error(e);
            }
        }).publishOn(Schedulers.boundedElastic()) // 切换到弹性线程池
            .doOnSubscribe(sub -> log.info("开始生成，sessionId={}", sessionId))
            .doOnComplete(() -> log.info("生成完成，sessionId={}", sessionId))
            .doOnError(e -> log.error("生成失败，sessionId={}", sessionId, e));
    }

    /**
     * 响应式保存消息
     */
    private Mono<Message> saveMessage(String sessionId, String messageId, String content) {
        return Mono.fromCallable(() -> {
            Message message = new Message();
            message.setMessageId(messageId);
            message.setSessionId(sessionId);
            message.setContent(content);
            message.setRole(MessageRole.ASSISTANT);
            return messageRepository.save(message);
        }).subscribeOn(Schedulers.boundedElastic());
    }
}

五、底层原理剖析

5.1 SseEmitter 底层原理

5.1.1 Servlet 异步机制

SseEmitter 基于 Servlet 3.0 的异步支持实现：

// Spring MVC 底层处理流程
public class SseEmitter extends ResponseBodyEmitter {
    public SseEmitter(Long timeout) {
        super();
        this.timeout = timeout;
    }

    @Override
    protected void extendResponse(ServerHttpResponse outputMessage) {
        super.extendResponse(outputMessage);
        // 设置 SSE 必需的响应头
        outputMessage.getHeaders().setContentType(MediaType.TEXT_EVENT_STREAM);
        outputMessage.getHeaders().setCacheControl(CacheControl.noCache());
    }
}

关键实现：

// ResponseBodyEmitter 核心实现
public abstract class ResponseBodyEmitter {
    private final Set<DataWithMediaType> earlySendAttempts = new LinkedHashSet<>(4);
    private Handler handler;
    private boolean complete;

    public void send(Object object) throws IOException {
        if (this.handler != null) {
            try {
                this.handler.send(object, null); // 直接写入响应流
            } catch (IOException ex) {
                throw ex;
            } catch (Throwable ex) {
                throw new IllegalStateException("Failed to send " + object, ex);
            }
        } else {
            // 请求还未初始化完成，先缓存
            this.earlySendAttempts.add(new DataWithMediaType(object, null));
        }
    }

    public void complete() {
        if (this.handler != null) {
            this.handler.complete();
        } else {
            this.complete = true;
        }
    }
}

5.1.2 异步 Servlet 工作流程

1. 客户端发起请求
2. Tomcat 接收请求，分配线程 A 处理
3. Controller 返回 SseEmitter
4. Spring MVC 调用 AsyncContext.startAsync()
5. 线程 A 释放，返回线程池
6. 业务逻辑在线程 B 中执行
7. 调用 emitter.send() 写入数据
8. 数据通过 AsyncContext 写入 TCP 缓冲区
9. Tomcat 的 Poller 线程监听 Socket 可写事件
10. 数据发送到客户端
11. emitter.complete() 关闭连接

核心代码（Spring 源码）：

// DeferredResultInterceptor.java
public class ResponseBodyEmitterReturnValueHandler implements HandlerMethodReturnValueHandler {
    @Override
    public void handleReturnValue(Object returnValue, ...) throws Exception {
        ResponseBodyEmitter emitter = (ResponseBodyEmitter) returnValue;
        // 启动异步上下文
        WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(webRequest);
        DeferredResult<?> deferredResult = new DeferredResult<>();
        asyncManager.startDeferredResultProcessing(deferredResult, ...);
        // 设置发送处理器
        emitter.initialize(new HttpMessageConvertingHandler(outputMessage, ...));
    }
}

5.1.3 数据发送流程

// 数据如何写入客户端
class HttpMessageConvertingHandler implements ResponseBodyEmitter.Handler {
    @Override
    public void send(Object data, MediaType mediaType) throws IOException {
        // 1. 序列化数据
        if (data instanceof String) {
            String text = (String) data;
            // SSE 格式：data: xxx\n\n
            String formattedData = "data: " + text + "\n\n";
            byte[] bytes = formattedData.getBytes(StandardCharsets.UTF_8);
            // 2. 写入 OutputStream
            this.outputMessage.getBody().write(bytes);
            // 3. flush 立即发送（重要！）
            this.outputMessage.getBody().flush();
        }
    }

    @Override
    public void complete() {
        // 关闭输出流
        this.outputMessage.getBody().close();
    }
}

为什么需要 flush()？

• HTTP 响应默认是缓冲的，只有 flush 才会立即发送到客户端
• SSE 的实时性依赖于每次 send 都立即 flush

5.2 Flux 底层原理

5.2.1 Reactive Streams 协议

Flux 实现了 Reactive Streams 规范，核心是背压控制：

// 完整的订阅流程
Flux<String> flux = Flux.just("A", "B", "C");
flux.subscribe(new Subscriber<String>() {
    private Subscription subscription;

    @Override
    public void onSubscribe(Subscription s) {
        this.subscription = s;
        // 请求 1 个元素（背压控制）
        s.request(1);
    }

    @Override
    public void onNext(String item) {
        System.out.println("收到：" + item);
        // 处理完后再请求下一个
        subscription.request(1);
    }

    @Override
    public void onError(Throwable t) {
        System.err.println("错误：" + t);
    }

    @Override
    public void onComplete() {
        System.out.println("完成");
    }
});

关键点：

• 消费者通过 request(n) 主动拉取数据，避免被淹没
• 生产者必须尊重请求数量，不能无限推送

5.2.2 操作符链式原理

// 每个操作符都返回一个新的 Flux，形成链式结构
Flux<Integer> flux = Flux.range(1, 10)
    .map(i -> i * 2)           // FluxMap
    .filter(i -> i > 5)        // FluxFilter
    .take(5);                  // FluxTake

// 实际结构
FluxTake(FluxFilter(FluxMap(FluxRange(1, 10))))

订阅传播：

订阅 (subscribe)：从最外层向内传播
FluxTake → FluxFilter → FluxMap → FluxRange

数据流 (onNext)：从最内层向外传播
FluxRange → FluxMap → FluxFilter → FluxTake → Subscriber

5.2.3 线程调度原理

// Schedulers 的本质是 Executor 包装
public abstract class Schedulers {
    // 立即执行（当前线程）
    static Scheduler immediate() { return ImmediateScheduler.INSTANCE; }
    // 单线程
    static Scheduler single() { return SingleScheduler.INSTANCE; }
    // 弹性线程池（IO 密集）
    static Scheduler boundedElastic() { return BoundedElasticScheduler.INSTANCE; }
    // 并行线程池（CPU 密集）
    static Scheduler parallel() { return ParallelScheduler.INSTANCE; }
}

// 线程切换实现
Flux.just("A")
    .publishOn(Schedulers.boundedElastic()) // 切换到弹性线程池
    .map(s -> {
        System.out.println("map 线程：" + Thread.currentThread().getName());
        return s.toLowerCase();
    })
    .subscribeOn(Schedulers.parallel()) // 订阅在并行线程池
    .subscribe();

subscribeOn vs publishOn：

subscribeOn：影响源头（订阅操作的线程）
publishOn：影响下游（后续操作符的线程）

Flux.just("A")
    .doOnNext(s -> log("1: " + Thread.currentThread().getName()))
    .publishOn(Schedulers.single())
    .doOnNext(s -> log("2: " + Thread.currentThread().getName()))
    .subscribeOn(Schedulers.parallel())
    .subscribe();

输出:
1: parallel-1 (受 subscribeOn 影响)
2: single-1 (受 publishOn 影响)

5.2.4 冷流 vs 热流

// 冷流（Cold）：每次订阅都重新执行
Flux<Integer> cold = Flux.range(1, 3).doOnSubscribe(s -> System.out.println("订阅了"));
cold.subscribe(i -> System.out.println("订阅者 1: " + i));
cold.subscribe(i -> System.out.println("订阅者 2: " + i));
// 输出：
// 订阅了
// 订阅者 1: 1
// 订阅者 1: 2
// 订阅者 1: 3
// 订阅了 (再次执行)
// 订阅者 2: 1
// 订阅者 2: 2
// 订阅者 2: 3

// 热流（Hot）：多个订阅者共享数据源
ConnectableFlux<Integer> hot = Flux.range(1, 3).doOnSubscribe(s -> System.out.println("订阅了")).publish();
hot.subscribe(i -> System.out.println("订阅者 1: " + i));
hot.subscribe(i -> System.out.println("订阅者 2: " + i));
hot.connect(); // 开始发射
// 输出：
// 订阅了 (只执行一次)
// 订阅者 1: 1
// 订阅者 2: 1
// 订阅者 1: 2
// 订阅者 2: 2
// 订阅者 1: 3
// 订阅者 2: 3

六、性能对比与选型

6.1 性能对比

6.1.1 吞吐量测试

测试场景：10000 个并发请求，每个请求返回 100 个数据块

// 测试代码
@State(Scope.Benchmark)
public class PerformanceTest {
    @Benchmark
    public void testSseEmitter(Blackhole blackhole) {
        // 模拟 SseEmitter
        SseEmitter emitter = new SseEmitter();
        for (int i = 0; i < 100; i++) {
            emitter.send("data" + i);
        }
        emitter.complete();
    }

    @Benchmark
    public void testFlux(Blackhole blackhole) {
        // 模拟 Flux
        Flux.range(0, 100).map(i -> "data" + i).subscribe(blackhole::consume);
    }
}

测试结果（JMH Benchmark）：

6.1.2 内存分析

SseEmitter 内存结构：
Thread Stack ~1MB (线程栈)
Response Buffer ~8KB (HTTP 响应缓冲)
Connection State ~40KB (连接状态)
总计：~1.05MB/连接

Flux 内存结构：
Subscription ~2KB (订阅对象)
Operator Chain ~5KB (操作符链)
Buffer ~3KB (可配置)
总计：~10KB/连接

C10K 问题：
SseEmitter: 10000 连接 = 10GB 内存
Flux: 10000 连接 = 100MB 内存

6.2 技术选型

6.2.1 选型决策树

需要流式响应？
├─ 否 → 使用普通 REST 接口
└─ 是 → 继续
   | 现有项目是 Spring MVC？
   ├─ 是 → 考虑 SseEmitter
   | │
   | │ 并发量 < 1000？
   | ├─ 是 → SseEmitter (简单易用)
   | └─ 否 → 考虑迁移到 Flux
   └─ 否（新项目） → Flux (性能更好)
      | 团队熟悉响应式编程？
      ├─ 是 → 直接使用 Flux
      └─ 否 → 先用 SseEmitter，逐步迁移

6.2.2 详细对比

6.2.3 实际案例选型

案例 1：企业内部管理系统

• 并发：<500
• 团队：传统 Java 团队
• 选择：SseEmitter
• 理由：易于理解和维护，性能够用

案例 2：对外开放的 AI 聊天 API

• 并发：10000+
• 团队：有响应式经验
• 选择：Flux
• 理由：高性能、低资源占用、完整的背压控制

案例 3：实时监控大屏

• 并发：<100
• 需求：推送服务器指标
• 选择：SseEmitter
• 理由：简单场景，快速实现

案例 4：物联网数据采集平台

• 并发：50000+设备
• 数据：高频率传感器数据
• 选择：Flux + R2DBC
• 理由：全栈响应式，极致性能

七、生产环境实战

7.1 完整项目改造

7.1.1 改造前（同步模式）

@RestController
@RequestMapping("/api/v1/chat")
public class ChatController {
    @PostMapping("/send")
    public Result<MessageVO> sendMessage(@RequestBody SendMessageReq req) {
        // 阻塞等待 AI 回复（可能 30 秒）
        String response = aiService.generate(req.getContent());
        return Result.success(new MessageVO(response));
    }
}

问题：

• 用户体验差：等待 30 秒
• 资源浪费：占用线程 30 秒
• 容易超时：Nginx/Gateway 60 秒超时

7.1.2 改造后（流式模式）

@RestController
@RequestMapping("/api/v1/chat")
@RequiredArgsConstructor
public class StreamChatController {
    private final StreamChatService chatService;

    /**
     * 流式发送消息（推荐）
     */
    @PostMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<ChatResponse>> streamMessage(@RequestBody SendMessageReq req) {
        return chatService.streamChat(req)
            .map(chunk -> ServerSentEvent.<ChatResponse>builder()
                .event("message")
                .id(chunk.getMessageId())
                .data(chunk)
                .build())
            .concatWith(completionEvent())
            .onErrorResume(this::handleError);
    }

    private Flux<ServerSentEvent<ChatResponse>> completionEvent() {
        return Flux.just(ServerSentEvent.<ChatResponse>builder()
            .event("complete")
            .data(ChatResponse.complete())
            .build());
    }

    private Flux<ServerSentEvent<ChatResponse>> handleError(Throwable e) {
        log.error("流式聊天异常", e);
        return Flux.just(ServerSentEvent.<ChatResponse>builder()
            .event("error")
            .data(ChatResponse.error(e.getMessage()))
            .build());
    }
}

7.2 Service 层实现

@Service
@Slf4j
@RequiredArgsConstructor
public class StreamChatService {
    private final OpenAIClient openAIClient;
    private final ConversationRepository conversationRepository;
    private final MessageRepository messageRepository;
    private final RedisTemplate<String, Object> redisTemplate;

    /**
     * 流式聊天核心逻辑
     */
    public Flux<ChatResponse> streamChat(SendMessageReq req) {
        String messageId = generateMessageId();
        String sessionId = req.getSessionId();
        return Flux.create(sink -> {
            StringBuilder fullResponse = new StringBuilder();
            AtomicInteger chunkCount = new AtomicInteger(0);
            try {
                // 1. 保存用户消息
                saveUserMessage(sessionId, req.getContent()).subscribe();
                // 2. 调用 OpenAI 流式 API
                openAIClient.streamChatCompletion(buildPrompt(sessionId, req.getContent()), new StreamCallback() {
                    @Override
                    public void onChunk(String chunk) {
                        fullResponse.append(chunk);
                        // 发送数据块
                        ChatResponse response = ChatResponse.builder()
                            .messageId(messageId)
                            .sessionId(sessionId)
                            .content(chunk)
                            .chunkIndex(chunkCount.getAndIncrement())
                            .isComplete(false)
                            .build();
                        sink.next(response);
                        // 缓存到 Redis（支持断线重连）
                        cacheChunk(messageId, chunkCount.get(), chunk);
                    }

                    @Override
                    public void onComplete() {
                        // 3. 保存完整 AI 回复
                        saveAssistantMessage(sessionId, messageId, fullResponse.toString())
                            .doOnSuccess(msg -> log.info("消息保存成功：{}", messageId))
                            .doOnError(e -> log.error("消息保存失败", e))
                            .subscribe();
                        // 4. 发送完成信号
                        ChatResponse finalResponse = ChatResponse.builder()
                            .messageId(messageId)
                            .sessionId(sessionId)
                            .content("")
                            .chunkIndex(chunkCount.get())
                            .isComplete(true)
                            .totalChunks(chunkCount.get())
                            .build();
                        sink.next(finalResponse);
                        sink.complete();
                        // 5. 清理缓存
                        clearCache(messageId);
                    }

                    @Override
                    public void onError(Throwable error) {
                        log.error("AI 生成失败", error);
                        sink.error(new BusinessException("AI 服务异常：" + error.getMessage()));
                    }
                });
            } catch (Exception e) {
                sink.error(e);
            }
        }).publishOn(Schedulers.boundedElastic())
            .timeout(Duration.ofMinutes(5)) // 5 分钟超时
            .doOnSubscribe(sub -> log.info("开始流式聊天，sessionId={}", sessionId))
            .doOnComplete(() -> log.info("流式聊天完成，sessionId={}", sessionId))
            .doOnError(e -> log.error("流式聊天异常，sessionId={}", sessionId, e));
    }

    /**
     * 保存用户消息（响应式）
     */
    private Mono<Message> saveUserMessage(String sessionId, String content) {
        return Mono.fromCallable(() -> {
            Message message = Message.builder()
                .messageId(generateMessageId())
                .sessionId(sessionId)
                .role(MessageRole.USER)
                .content(content)
                .createTime(LocalDateTime.now())
                .build();
            return messageRepository.save(message);
        }).subscribeOn(Schedulers.boundedElastic());
    }

    /**
     * 保存 AI 回复消息（响应式）
     */
    private Mono<Message> saveAssistantMessage(String sessionId, String messageId, String content) {
        return Mono.fromCallable(() -> {
            Message message = Message.builder()
                .messageId(messageId)
                .sessionId(sessionId)
                .role(MessageRole.ASSISTANT)
                .content(content)
                .createTime(LocalDateTime.now())
                .build();
            return messageRepository.save(message);
        }).subscribeOn(Schedulers.boundedElastic());
    }

    /**
     * 缓存数据块（支持断线重连）
     */
    private void cacheChunk(String messageId, int index, String chunk) {
        String key = "chat:stream:" + messageId;
        redisTemplate.opsForList().rightPush(key, chunk);
        redisTemplate.expire(key, Duration.ofMinutes(10));
    }

    /**
     * 清理缓存
     */
    private void clearCache(String messageId) {
        String key = "chat:stream:" + messageId;
        redisTemplate.delete(key);
    }

    private String generateMessageId() {
        return "msg_" + System.currentTimeMillis() + "_" + RandomUtil.randomString(8);
    }
}

7.3 前端对接实现

7.3.1 原生 JavaScript

class StreamChatClient {
    constructor(apiBaseUrl) {
        this.apiBaseUrl = apiBaseUrl;
        this.eventSource = null;
    }

    /**
     * 发送流式消息
     */
    sendStreamMessage(sessionId, content, callbacks) {
        const url = `${this.apiBaseUrl}/api/v1/chat/stream`;
        // 使用 fetch 进行 POST 请求，获取 ReadableStream
        fetch(url, {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json',
                'Accept': 'text/event-stream'
            },
            body: JSON.stringify({
                sessionId: sessionId,
                content: content
            })
        }).then(response => {
            const reader = response.body.getReader();
            const decoder = new TextDecoder();
            // 读取流
            const readChunk = () => {
                reader.read().then(({ done, value }) => {
                    if (done) {
                        callbacks.onComplete?.();
                        return;
                    }
                    // 解析 SSE 数据
                    const chunk = decoder.decode(value);
                    const lines = chunk.split('\n');
                    let eventType = 'message';
                    let data = '';
                    for (const line of lines) {
                        if (line.startsWith('event:')) {
                            eventType = line.substring(6).trim();
                        } else if (line.startsWith('data:')) {
                            data = line.substring(5).trim();
                        } else if (line === '' && data) {
                            // 完整的事件
                            this.handleEvent(eventType, data, callbacks);
                            eventType = 'message';
                            data = '';
                        }
                    }
                    // 继续读取
                    readChunk();
                });
            };
            readChunk();
        }).catch(error => {
            console.error('Stream error:', error);
            callbacks.onError?.(error);
        });
    }

    /**
     * 处理 SSE 事件
     */
    handleEvent(eventType, data, callbacks) {
        try {
            const parsed = JSON.parse(data);
            switch (eventType) {
                case 'message':
                    callbacks.onMessage?.(parsed.content);
                    break;
                case 'complete':
                    callbacks.onComplete?.();
                    break;
                case 'error':
                    callbacks.onError?.(new Error(parsed.message));
                    break;
            }
        } catch (e) {
            console.error('Parse error:', e);
        }
    }
}

// 使用示例
const client = new StreamChatClient('http://localhost:8080');
client.sendStreamMessage('sess_123', '你好，介绍一下你自己', {
    onMessage: (content) => {
        // 逐字显示
        document.getElementById('response').innerText += content;
    },
    onComplete: () => {
        console.log('流式响应完成');
    },
    onError: (error) => {
        console.error('错误:', error);
        alert('发生错误：' + error.message);
    }
});

7.3.2 React 实现

import React, { useState, useEffect, useRef } from 'react';

interface ChatMessage {
    role: 'user' | 'assistant';
    content: string;
    isStreaming?: boolean;
}

export const StreamChat: React.FC = () => {
    const [messages, setMessages] = useState<ChatMessage[]>([]);
    const [inputValue, setInputValue] = useState('');
    const [isLoading, setIsLoading] = useState(false);
    const messagesEndRef = useRef<HTMLDivElement>(null);

    // 自动滚动到底部
    useEffect(() => {
        messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' });
    }, [messages]);

    const sendMessage = async () => {
        if (!inputValue.trim() || isLoading) return;
        const userMessage = inputValue;
        setInputValue('');
        setIsLoading(true);

        // 添加用户消息
        setMessages(prev => [...prev, { role: 'user', content: userMessage }]);

        // 添加 AI 消息占位符
        const aiMessageIndex = messages.length + 1;
        setMessages(prev => [...prev, { role: 'assistant', content: '', isStreaming: true }]);

        try {
            const response = await fetch('/api/v1/chat/stream', {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/json',
                    'Accept': 'text/event-stream'
                },
                body: JSON.stringify({
                    sessionId: 'sess_123',
                    content: userMessage
                })
            });
            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.startsWith('data:')) {
                        const data = line.substring(5).trim();
                        if (data) {
                            try {
                                const parsed = JSON.parse(data);
                                if (parsed.content) {
                                    // 更新 AI 消息内容
                                    setMessages(prev => {
                                        const newMessages = [...prev];
                                        newMessages[aiMessageIndex] = {
                                            ...newMessages[aiMessageIndex],
                                            content: newMessages[aiMessageIndex].content + parsed.content
                                        };
                                        return newMessages;
                                    });
                                }
                                if (parsed.isComplete) {
                                    // 标记流结束
                                    setMessages(prev => {
                                        const newMessages = [...prev];
                                        newMessages[aiMessageIndex].isStreaming = false;
                                        return newMessages;
                                    });
                                }
                            } catch (e) {
                                console.error('Parse error:', e);
                            }
                        }
                    }
                }
            }
        } catch (error) {
            console.error('Stream error:', error);
            alert('发生错误：' + error);
        } finally {
            setIsLoading(false);
        }
    };

    return (
        <div className="chat-container">
            <div className="messages">
                {messages.map((msg, index) => (
                    <div key={index} className={`message message-${msg.role}`}>
                        <div className="message-content">
                            {msg.content}
                            {msg.isStreaming && <span className="cursor">▊</span>}
                        </div>
                    </div>
                ))}
                <div ref={messagesEndRef} />
            </div>
            <div className="input-area">
                <input
                    type="text"
                    value={inputValue}
                    onChange={(e) => setInputValue(e.target.value)}
                    onKeyPress={(e) => e.key === 'Enter' && sendMessage()}
                    placeholder="输入消息..."
                    disabled={isLoading}
                />
                <button onClick= =>
                    {isLoading ? '发送中...' : '发送'}
                
            
        
    );
};

7.4 异常处理与监控

7.4.1 超时处理

@Configuration
public class WebFluxConfig {
    @Bean
    public WebClient webClient() {
        return WebClient.builder()
            .clientConnector(new ReactorClientHttpConnector(
                HttpClient.create()
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
                    .responseTimeout(Duration.ofMinutes(5))
                    .doOnConnected(conn -> conn
                        .addHandlerLast(new ReadTimeoutHandler(300))
                        .addHandlerLast(new WriteTimeoutHandler(300))
                    )
            ))
            .build();
    }
}

// Flux 超时处理
public Flux<String> streamWithTimeout() {
    return Flux.create(sink -> {
        // 流式逻辑
    }).timeout(Duration.ofMinutes(5), Flux.just("超时兜底数据"))
        .onErrorResume(TimeoutException.class, e -> {
            log.error("流式处理超时", e);
            return Flux.just("处理时间过长，请稍后重试");
        });
}

7.4.2 监控埋点

@Aspect
@Component
@Slf4j
public class StreamMonitorAspect {
    @Around("@annotation(streamMonitor)")
    public Object monitor(ProceedingJoinPoint pjp, StreamMonitor streamMonitor) throws Throwable {
        String methodName = pjp.getSignature().getName();
        long startTime = System.currentTimeMillis();
        Object result = pjp.proceed();
        if (result instanceof Flux) {
            Flux<?> flux = (Flux<?>) result;
            AtomicLong chunkCount = new AtomicLong(0);
            AtomicLong totalBytes = new AtomicLong(0);
            return flux
                .doOnNext(item -> {
                    chunkCount.incrementAndGet();
                    if (item instanceof String) {
                        totalBytes.addAndGet(((String) item).length());
                    }
                })
                .doOnComplete(() -> {
                    long cost = System.currentTimeMillis() - startTime;
                    log.info("流式方法执行完成：method={}, chunks={}, bytes={}, cost={}ms", methodName, chunkCount.get(), totalBytes.get(), cost);
                    // 上报监控指标
                    MetricsCollector.recordStreamMetrics(methodName, chunkCount.get(), totalBytes.get(), cost);
                })
                .doOnError(error -> {
                    log.error("流式方法执行失败：method={}, error={}", methodName, error.getMessage(), error);
                    // 上报错误
                    MetricsCollector.recordStreamError(methodName, error);
                });
        }
        return result;
    }
}

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface StreamMonitor {
    String value() default "";
}

// 使用
@StreamMonitor("chat-stream")
public Flux<ChatResponse> streamChat(SendMessageReq req) {
    // ...
}

八、常见问题与解决方案

8.1 SseEmitter 常见问题

Q1: SseEmitter 连接意外断开

现象： 客户端随机断开连接，无错误日志

原因：

1. Nginx/Gateway 超时配置
2. 网络不稳定
3. 浏览器 Tab 切换（移动端）

解决方案：

# Nginx 配置
location /api/ {
    proxy_pass http://backend;
    # SSE 必需配置
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    chunked_transfer_encoding off;
    proxy_buffering off;
    proxy_cache off;
    # 超时设置
    proxy_connect_timeout 10s;
    proxy_send_timeout 600s; # 10 分钟
    proxy_read_timeout 600s; # 10 分钟
}

// 心跳保活
@Scheduled(fixedRate = 30000) // 每 30 秒
public void sendHeartbeat() {
    emitterManager.getAllEmitters().forEach(emitter -> {
        try {
            emitter.send(SseEmitter.event().name("heartbeat").data("ping"));
        } catch (IOException e) {
            log.warn("发送心跳失败，移除连接", e);
            emitterManager.remove(emitter);
        }
    });
}

Q2: 内存泄漏

现象： 服务器内存持续增长，最终 OOM

原因： SseEmitter 未正确关闭，导致连接泄漏

解决方案：

@Component
public class SseEmitterManager {
    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();

    public SseEmitter create(String sessionId, Long timeout) {
        SseEmitter emitter = new SseEmitter(timeout);
        // 完成时移除
        emitter.onCompletion(() -> {
            log.info("SSE 连接完成：{}", sessionId);
            emitters.remove(sessionId);
        });
        // 超时时移除
        emitter.onTimeout(() -> {
            log.warn("SSE 连接超时：{}", sessionId);
            emitters.remove(sessionId);
            try {
                emitter.complete();
            } catch (Exception e) {
                log.error("关闭超时连接失败", e);
            }
        });
        // 错误时移除
        emitter.onError(throwable -> {
            log.error("SSE 连接异常：{}", sessionId, throwable);
            emitters.remove(sessionId);
        });
        emitters.put(sessionId, emitter);
        return emitter;
    }

    // 定期清理过期连接
    @Scheduled(fixedRate = 60000) // 每分钟
    public void cleanup() {
        long now = System.currentTimeMillis();
        emitters.entrySet().removeIf(entry -> {
            // 实现自定义清理逻辑
            return false;
        });
    }
}

Q3: 数据丢失

现象： 客户端收到的数据不完整

原因： 缓冲区未及时 flush

解决方案：

public void send(String data) throws IOException {
    emitter.send(SseEmitter.event().data(data));
    // SseEmitter 内部会自动 flush，但如果自定义实现需要注意：
    // outputStream.write(data.getBytes());
    // outputStream.flush(); // 必须！
}

8.2 Flux 常见问题

Q1: 背压溢出

现象： reactor.core.Exceptions$OverflowException: Queue is full

原因： 生产速度 > 消费速度，且缓冲区满

解决方案：

Flux<String> flux = Flux.create(sink -> {
    // 生产逻辑
}, FluxSink.OverflowStrategy.BUFFER) // 或 DROP、LATEST、ERROR
    .onBackpressureBuffer(1000, dropped -> log.warn("丢弃数据：{}", dropped));

Q2: 线程阻塞

现象： 响应式代码仍然很慢

原因： 在响应式流中使用了阻塞操作

错误示例：

Flux.create(sink -> {
    String result = blockingHttpClient.get(); // 阻塞！
    sink.next(result);
    sink.complete();
})

正确做法：

Flux.create(sink -> {
    // 使用异步客户端
    asyncHttpClient.get().thenAccept(result -> {
        sink.next(result);
        sink.complete();
    }).exceptionally(error -> {
        sink.error(error);
        return null;
    });
})
// 或使用 subscribeOn 切换线程
.subscribeOn(Schedulers.boundedElastic())

Q3: 订阅未触发

现象： Flux 定义了但没有执行

原因： Flux 是惰性的，只有订阅才会执行

错误示例：

@GetMapping("/test")
public void test() {
    Flux.range(1, 10).map(i -> i * 2).doOnNext(System.out::println); // 没有 subscribe，不会执行！
}

正确做法：

@GetMapping("/test")
public Flux<Integer> test() {
    return Flux.range(1, 10).map(i -> i * 2); // Spring WebFlux 会自动订阅
}
// 或手动订阅
flux.subscribe(
    data -> System.out.println(data),
    error -> System.err.println(error),
    () -> System.out.println("完成")
);

8.3 生产环境最佳实践

1. 设置合理的超时时间

// 不要设置为 0（永不超时）
SseEmitter emitter = new SseEmitter(5 * 60 * 1000L); // 5 分钟
// Flux 也要设置超时
flux.timeout(Duration.ofMinutes(5))

2. 限制并发连接数

@Component
public class ConnectionLimiter {
    private final Semaphore semaphore = new Semaphore(1000); // 最多 1000 个连接

    public SseEmitter createWithLimit() throws InterruptedException {
        if (!semaphore.tryAcquire(5, TimeUnit.SECONDS)) {
            throw new BusinessException("服务器繁忙，请稍后重试");
        }
        SseEmitter emitter = new SseEmitter();
        emitter.onCompletion(() -> semaphore.release());
        emitter.onTimeout(() -> semaphore.release());
        emitter.onError(e -> semaphore.release());
        return emitter;
    }
}

3. 日志与监控

Flux<String> flux = Flux.create(sink -> {
    // 业务逻辑
}).doOnSubscribe(sub -> {
    log.info("开始流式处理：{}", contextInfo);
    MetricsCollector.incrementActiveStreams();
}).doOnNext(item -> {
    log.debug("发送数据块：{}", item);
    MetricsCollector.recordChunk();
}).doOnComplete(() -> {
    log.info("流式处理完成：{}", contextInfo);
    MetricsCollector.decrementActiveStreams();
}).doOnError(error -> {
    log.error("流式处理失败：{}", contextInfo, error);
    MetricsCollector.recordError();
    MetricsCollector.decrementActiveStreams();
});

4. 优雅关闭

@Component
public class GracefulShutdown implements ApplicationListener<ContextClosedEvent> {
    @Autowired
    private SseEmitterManager emitterManager;

    @Override
    public void onApplicationEvent(ContextClosedEvent event) {
        log.info("应用关闭，断开所有 SSE 连接");
        emitterManager.getAllEmitters().forEach(emitter -> {
            try {
                emitter.send(SseEmitter.event().name("shutdown").data("服务器即将重启，请重新连接"));
                emitter.complete();
            } catch (IOException e) {
                log.error("发送关闭通知失败", e);
            }
        });
    }
}

总结

核心要点

1. SseEmitter：
   • 基于 Servlet 异步，简单易用
   • 适合中小型项目（并发<1000）
   • 每个连接占用一个线程，资源开销较大
   • 不支持背压控制
2. Flux：
   • 基于 Reactive Streams，性能卓越
   • 适合高并发场景（并发>1000）
   • 异步非阻塞，资源利用率高
   • 完整的背压控制和错误处理
3. 选型建议：
   • 快速上手、团队经验不足 → SseEmitter
   • 高性能、大规模并发 → Flux
   • 新项目推荐直接使用 Flux
   • 老项目可以先用 SseEmitter，逐步迁移

未来趋势

响应式编程已成为 Java 生态的重要方向，Spring、R2DBC、Kafka、Redis 等主流框架都已支持响应式。掌握 Flux 不仅能提升系统性能，也是技术成长的必经之路。