Java响应式编程之Flux与SseEmitter深度解析(附详细代码)

@PostMapping("/chat/message")
public Result<MessageVO> sendMessage(@RequestBody SendMessageReq req) {
    // 调用AI服务生成回复（可能需要5-30秒）
    String aiResponse = aiService.generateResponse(req.getContent());
    return Result.success(new MessageVO(aiResponse));
}

传统模式：
客户端 ----请求----> 服务端
                    [等待30秒]
客户端 <---完整响应-- 服务端

流式模式：
客户端 ----请求----> 服务端
客户端 <---数据块1--- 服务端 (0.5秒)
客户端 <---数据块2--- 服务端 (1.0秒)
客户端 <---数据块3--- 服务端 (1.5秒)
...
客户端 <---完成----- 服务端 (30秒)

data: 这是第一条消息\n\n
data: 这是第二条消息\n\n
event: custom-event\n
data: {"message": "JSON数据"}\n
id: 123\n\n

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

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

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

@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;
    }
}

// 原生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();
});

@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();
    }
}

@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() // 拒绝策略
        );
    }
}

// 服务端发送不同类型的事件
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);
});

@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请求头

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

@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("流完成"));
    }
}

@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;
}

// 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

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

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);

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)) // 指数退避重试
);

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();

@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());
    }
}

// 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;
        }
    }
}

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()关闭连接

// 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, ...));
    }
}

// 数据如何写入客户端
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();
    }
}

// 完整的订阅流程
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("完成");
    }
});

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

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

// 简化的FluxMap实现
class FluxMap<T, R> extends Flux<R> {

    private final Flux<T> source;
    private final Function<T, R> mapper;

    @Override
    public void subscribe(Subscriber<? super R> actual) {
        // 创建包装订阅者
        source.subscribe(new MapSubscriber<>(actual, mapper));
    }

    static class MapSubscriber<T, R> implements Subscriber<T> {
        private final Subscriber<? super R> actual;
        private final Function<T, R> mapper;

        @Override
        public void onNext(T item) {
            R mapped = mapper.apply(item); // 转换
            actual.onNext(mapped);         // 传递给下游
        }

        // ... 其他方法
    }
}

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

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

// 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：影响源头（订阅操作的线程）
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影响)

// 冷流（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

// 测试代码
@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);
    }
}

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内存

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

@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));
    }
}

@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());
    }
}

@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);
    }
}

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);
    }
});

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={sendMessage} disabled={isLoading}>
                    {isLoading ? '发送中...' : '发送'}
                </button>
            </div>
        </div>
    );
};

@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("处理时间过长，请稍后重试");
    });
}

@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) {
    // ...
}

# 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);
        }
    });
}

@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;
        });
    }
}

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

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

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())

@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("完成")
);

// 不要设置为0（永不超时）
SseEmitter emitter = new SseEmitter(5 * 60 * 1000L); // 5分钟

// Flux也要设置超时
flux.timeout(Duration.ofMinutes(5))

@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;
    }
}

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();
});

@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);
            }
        });
    }
}