DeepSeekApi对接流式输出异步聊天功能：基于Spring Boot和OkHttp的SSE应用实现

实现异步聊天功能：基于Spring Boot和OkHttp的SSE应用

在现代Web应用程序开发中，实时更新的能力对于增强用户体验至关重要。本文将详细介绍如何利用Spring Boot框架结合OkHttp库实现一个简单的异步聊天服务，该服务能够接收用户输入并通过Server-Sent Events (SSE) 向客户端推送实时更新。

一、技术栈选择

为了构建这个服务，我们选择了以下技术：

Spring Boot：用于快速搭建RESTful API服务。
OkHttp：一款高效的HTTP客户端库，适用于Java应用，支持异步请求处理。
SSE(Server-Sent Events)：一种允许服务器向浏览器推送更新的技术，特别适合于需要实时更新的应用场景。

二、核心逻辑详解

ChatController 类概览

我们的ChatController类主要包含了一个名为oneShot的方法，它负责接收POST请求，处理请求，并通过SSE返回结果给客户端。以下是该方法的具体实现细节。

定义HttpClient实例

首先，我们定义了一个OkHttpClient实例，设置了读取、写入和连接超时时间为60秒，以确保在网络状况不佳时仍有足够的时间完成操作。

private final OkHttpClient client = new OkHttpClient.Builder().readTimeout(60, TimeUnit.SECONDS).writeTimeout(60, TimeUnit.SECONDS).connectTimeout(60, TimeUnit.SECONDS).build();

oneShot 方法

oneShot方法是一个POST请求映射方法，它接收来自客户端的JSON格式数据，并转发至另一个API进行处理，最后通过SSE将结果推送给客户端。

@PostMapping(value = "/chat/oneShot", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter oneShot(@RequestBody String requestBody) {SseEmitter emitter = new SseEmitter(); // 创建一个新的SseEmitter实例用于异步事件流传输。// 使用OkHttp创建请求体，设置媒体类型为application/json，内容为requestBody。RequestBody body = RequestBody.create(MediaType.APPLICATION_JSON_VALUE, requestBody);// 构造一个HTTP POST请求，目标URL为http://10.155.28.19:11434/api/chat，携带body作为请求体。Request request = new Request.Builder().url("http://DeepSeek地址").post(body).build();// 异步执行HTTP请求，并注册回调处理响应或错误。client.newCall(request).enqueue(new Callback() {@Overridepublic void onFailure(Call call, IOException e) { // 请求失败时调用此方法。handleError(emitter, e); // 调用handleError处理错误。}@Overridepublic void onResponse(Call call, Response response) throws IOException { // 请求成功时调用此方法。if (!response.isSuccessful()) { // 检查响应是否成功。handleError(emitter, new IOException("Unexpected code " + response)); // 如果不成功，则处理错误。return;}try (ResponseBody responseBody = response.body()) { // 获取响应体。if (responseBody != null) { // 确保响应体非空。processResponseStream(responseBody.byteStream(), emitter); // 处理响应流。}} catch (Exception e) { // 捕获可能的异常。handleError(emitter, e); // 出现异常时处理错误。}}});return emitter; // 返回SseEmitter实例，开始向客户端发送事件流。
}

错误处理方法

当发生网络错误或其他异常情况时，我们通过handleError方法将错误信息发送给客户端。

private void handleError(SseEmitter emitter, Exception e) {try {emitter.send(SseEmitter.event().id("error").data(e.getMessage())); // 发送错误事件。emitter.completeWithError(e); // 完成SseEmitter，并附带错误。} catch (IOException ex) { // 捕获可能的IO异常。ex.printStackTrace(); // 打印堆栈跟踪。}
}

响应流处理方法

该方法负责从响应流中读取数据，并将其分割成单独的JSON对象，然后通过SSE发送给客户端。

private void processResponseStream(java.io.InputStream inputStream, SseEmitter emitter) {try {byte[] buffer = new byte[1024]; // 创建一个1024字节的缓冲区。int bytesRead; // 存储每次读取的字节数。while ((bytesRead = inputStream.read(buffer)) != -1) { // 循环读取输入流直到结束。String data = new String(buffer, 0, bytesRead); // 将读取的字节转换为字符串。// 假设每条消息是以换行符分隔的JSON对象。for (String line : data.split("\n")) { // 分割字符串，按行处理。if (!line.trim().isEmpty()) { // 忽略空行。emitter.send(SseEmitter.event().data(line)); // 发送每一行数据作为事件。}}}emitter.complete(); // 完成SseEmitter，表示所有数据已发送完毕。} catch (IOException e) { // 捕获可能的IO异常。handleError(emitter, e); // 出现异常时处理错误。}
}