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() {
        @Override
        public void onFailure(Call call, IOException e) { // 请求失败时调用此方法。
            handleError(emitter, e); // 调用handleError处理错误。
        }

        @Override
        public 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); // 出现异常时处理错误。
    }
}