Java后端SSE流式响应

场景通过调用wps团队问答接口将响应结果通过流式响应给前端controller层接受前端传递的问题与sessionId返回SseEmitter给前端Operation(summary 团队问答) PostMapping(value /searchGptSSE, produces text/event-stream;charsetUTF-8) public SseEmitter searchGptSSE(Validated(CloudDocumentReqVO.GptGroup.class) RequestBody CloudDocumentReqVO reqVO) { log.info([云文档-智能团队问答] 收到请求); Long userId SecurityFrameworkUtils.getLoginUserId(); return cloudDocumentService.searchGptSSE(userId,reqVO); }serviceImpl层获取用户token并执行wps接口调用逻辑Override public SseEmitter searchGptSSE(Long userId, CloudDocumentReqVO reqVO) { // 1. 获取用户access_token从Redis/DB读取 String accessToken wpsAuthService.getAccessTokenFormDBorRedis(userId); if (!StringUtils.hasText(accessToken)) { log.error([getDocLibrary] 用户未绑定WPS token或refreshToken已过期无法团队智能问答userId{}, userId); throw exception(WpsErrorCodeConstants.WPS_AUTH_REFRESH_TOKEN_IS_EXPTRED); } // 3.调用WPS API 团队问答 SseEmitter emitter null; try { emitter wpsApiClient.searchGpt(reqVO,accessToken); log.info([searchGpt] 团队问答 成功); } catch (Exception e) { log.error([searchGpt] 团队问答 失败, e); throw exception(WpsErrorCodeConstants.WPS_AI_SEARCH_GPT_ERROR, e); }finally { // 保存sessionId记录 saveWpsSessionIfAbsent(reqVO.getSessionId(),reqVO.getSearchName(),userId.toString()); } return emitter; }调用接口发送Http请求逻辑处理接口参数并执行请求获取响应ssemitter读取。public SseEmitter searchGpt(CloudDocumentReqVO reqVO, String accessToken) { //创建sseEmitter并设置连接超时10分钟 SseEmitter emitter new SseEmitter(10*60*1000L); try { // 2. 构建请求 URL String aiGptUrl WpsApiConstants.AI_SEARCH_GPT_URL; String baseAndCreatUrl buildBaseAuthUrl(aiGptUrl); // String finalUrl buildQueryUrl(baseAndCreatUrl, reqVO); String contentType WpsApiConstants.HEADER_CONTENT_TYPE_JSON; // 3. 生成 RFC1123 格式日期X-Kso-Date String xKsoDate getRfc1123Date(); // 4. 构造请求体 String json JSONUtil.toJsonStr(reqVO); byte[] requestBodyjson.getBytes(StandardCharsets.UTF_8); //byte[] requestBody new byte[0]; // 5. 计算 KSO-1 签名 KsoSign kso new KsoSign(properties.getClientId(), properties.getClientSecret()); KsoSign.Out signOut kso.kso1Sign(POST, aiGptUrl, contentType, xKsoDate, requestBody); System.out.println(kso.Date:signOut.getDate()); System.out.println(kso.Authorization:signOut.getAuthorization()); HttpRequest httpRequest HttpRequest.post(baseAndCreatUrl) .header(WpsApiConstants.CONTENT_TYPE, contentType) .header(WpsApiConstants.X_KSO_DATE, xKsoDate) .header(WpsApiConstants.X_KSO_AUTHORIZATION, signOut.getAuthorization()) .header(WpsApiConstants.AUTHORIZATION, WpsApiConstants.OAUTH_BEARER accessToken) .body(requestBody) .timeout(120000); // 记录开始时间 long start System.currentTimeMillis(); // 异步线程读取流推送SSEThreadUtilHutool中操作线程的工具execAsync异步 -{}:lambda表达式异步执行的业务逻辑 ThreadUtil.execAsync(() - { // Hutool HttpResponse实现AutoCloseable自动释放连接 try ( //发送异步Http请求不占用主线程阻塞底层通过线程池发送网络请求 HttpResponse resp httpRequest.executeAsync(); //获取字节输入流 InputStream inputStream resp.bodyStream(); //InputStreamReader将字节转为字符的桥梁指定UTF8编码默认编码容易中英文乱码BufferedReader带缓冲区的字符读取器 BufferedReader reader new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8))) { String line; while ((line reader.readLine()) ! null) { //按行读取字符内容 if(line!null line.startsWith(data:)){ line line.substring(5); } //读取一行像前端发送一行事件内容SSE是单向通道直接服务端发送不能前端传输。双向通信使用WebSocket emitter.send(SseEmitter.event().data(line)); } //所有数据推送完毕关闭当前Sse长连接 emitter.complete(); } catch (Exception e) { //异常中断 emitter.completeWithError(e); } }); // 计算耗时 long cost System.currentTimeMillis() - start; System.out.println(执行耗时111 cost ms); } catch (Exception e) { emitter.completeWithError(e); } return emitter; }注如果不用 try-with-resources发送Http请求和获取字节流需要手动关闭以下// 资源必须关闭避免连接泄漏 reader.close(); inputStream.close();