ResponseBodyEmitter 使用指南

最近在写ai返回的时候,使用到了这个工具类 在这里记录一下

在现代Web开发中，尤其是在处理长时间运行的请求时，如何有效地管理和返回响应成为了一个重要的话题。在此背景下，Spring框架提供了 ResponseBodyEmitter 类，以支持异步和基于事件的响应处理。

什么是 ResponseBodyEmitter？

ResponseBodyEmitter 是 Spring 提供的一个类，允许在处理请求时逐步发送数据。这种方式特别适合于实时数据传输或长时间运行的操作，如流式数据、服务器推送事件（Server-Sent Events, SSE）等。

主要特点

• 异步支持：可以在处理请求时不阻塞服务器线程，从而提高应用的响应能力。

• 逐步响应：允许在请求处理的过程中逐步发送数据，而不是等待所有数据准备好后再发送。

• 事件触发：可以根据特定事件触发响应的发送。

如何使用 ResponseBodyEmitter

1. 创建 Emitter

首先，你需要在控制器中创建一个 ResponseBodyEmitter 实例。例如：

java复制代码

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;

@RestController
public class MyController {

    @GetMapping("/stream")
    public ResponseBodyEmitter handle() {
        ResponseBodyEmitter emitter = new ResponseBodyEmitter();
        // 这里可以添加数据处理逻辑
        return emitter;
    }
}

2. 发送数据

在处理请求的过程中，可以通过 ResponseBodyEmitter 的 send() 方法逐步发送数据。例如：

java复制代码

import java.util.concurrent.Executors;

@GetMapping("/stream")
public ResponseBodyEmitter handle() {
    ResponseBodyEmitter emitter = new ResponseBodyEmitter();
    
    Executors.newSingleThreadExecutor().submit(() -> {
        try {
            emitter.send("数据片段1");
            Thread.sleep(1000); // 模拟延迟
            emitter.send("数据片段2");
            Thread.sleep(1000);
            emitter.send("数据片段3");
            emitter.complete(); // 完成响应
        } catch (Exception e) {
            emitter.completeWithError(e); // 处理异常
        }
    });

    return emitter;
}

3. 处理异常和完成

使用 complete() 方法来标记响应已完成，使用 completeWithError() 方法来处理异常情况。

实际使用示例

ResponseBodyEmitter emitter = new ResponseBodyEmitter();
        emitter.onCompletion(() -> {
            logger.info("开始使用gpt模型进行问答");
        });
        emitter.onError(throwable -> logger.error(throwable.getMessage()));
        Flux<ChatResponse> res = chatModel.stream(Prompt.builder().messages(UserMessage.builder().text("讲个笑话").build(),systemMessage)
                .build());
        res.subscribe(chatResponse -> {
                    if(Objects.nonNull(chatResponse.getResult().getOutput().getText())){
                        try {
                            emitter.send(chatResponse.getResult().getOutput().getText());
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    }

                },
                error -> {
                    logger.info(error.getMessage());
                },
                emitter::complete


        );

        return emitter;

注意事项

• 超时设置：可以设置 ResponseBodyEmitter 的超时时间，防止长时间未响应造成连接问题。

• 资源管理：确保在完成响应后正确释放资源，以避免内存泄漏。

• 线程安全：确保在多线程环境下对 ResponseBodyEmitter 的调用是线程安全的。

总结

ResponseBodyEmitter 是一个强大且灵活的工具，适用于需要实时数据传输的场景。通过合理的使用，可以显著提升应用的性能与用户体验。