peanut/happy-life-star
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 优化 SSE 转发逻辑，使用严格行读取和显式 flush

Browse Source
This commit is contained in:
Peanut
2026-07-21 23:28:00 +08:00
parent 2b6dc72744
commit 406f9b77b4
1 changed files with 26 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/main/java/com/emotion/service/impl/ShortNovelServiceImpl.java
+26 -5
View File
@@ -21,6 +21,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.EOFException;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
@@ -100,7 +101,7 @@ public class ShortNovelServiceImpl implements ShortNovelService {
upstreamBody.put("action", request.getAction()); upstreamBody.put("action", request.getAction());
upstreamBody.put("payload", request.getPayload()); upstreamBody.put("payload", request.getPayload());
return forwardSse("/api/novels/daily/conversation/stream", upstreamBody, currentUserId, null); return forwardSse("/api/novels/daily/conversation/stream", upstreamBody, currentUserId, request.getOriginalQuery());
} }
/** /**
@@ -139,12 +140,18 @@ public class ShortNovelServiceImpl implements ShortNovelService {
// source() 返回的 BufferedSource 逐行从网络 socket 读取，不会缓冲整个响应 // source() 返回的 BufferedSource 逐行从网络 socket 读取，不会缓冲整个响应
BufferedSource source = responseBody.source(); BufferedSource source = responseBody.source();
log.info("[ShortNovel SSE] 开始读取上游响应: path={}, userId={}", path, currentUserId);
StringBuilder dataBuffer = new StringBuilder(); StringBuilder dataBuffer = new StringBuilder();
while (!source.exhausted()) { while (!source.exhausted()) {
String line = source.readUtf8Line(); String line;
if (line == null) break; try {
line = source.readUtf8LineStrict();
} catch (EOFException e) {
break;
}
log.debug("[ShortNovel SSE] 读取到行: length={}, timestamp={}",
line.length(), System.currentTimeMillis());
if (line.startsWith("data:")) { if (line.startsWith("data:")) {
dataBuffer.append(line.substring(5).trim()); dataBuffer.append(line.substring(5).trim());
} else if (line.isEmpty() && dataBuffer.length() > 0) { } else if (line.isEmpty() && dataBuffer.length() > 0) {
@@ -158,17 +165,23 @@ public class ShortNovelServiceImpl implements ShortNovelService {
try { try {
JSONObject event = JSON.parseObject(dataStr); JSONObject event = JSON.parseObject(dataStr);
String type = event.getString("type"); String type = event.getString("type");
long eventTimestamp = System.currentTimeMillis();
log.info("[ShortNovel SSE] 处理事件: type={}, timestamp={}", type, eventTimestamp);
// 拦截 novel_done 事件，保存到数据库 // 拦截 novel_done 事件，保存到数据库
if ("novel_done".equals(type)) { if ("novel_done".equals(type)) {
JSONObject payload = event.getJSONObject("payload"); JSONObject payload = event.getJSONObject("payload");
if (payload != null && originalQuery != null) { log.info("[ShortNovel SSE] novel_done 事件: originalQuery={}, payload={}",
originalQuery, payload != null);
if (payload != null && originalQuery != null && !originalQuery.trim().isEmpty()) {
String fullText = payload.getString("full_text"); String fullText = payload.getString("full_text");
if (fullText != null) { if (fullText != null) {
Map<String, Object> metadata = new HashMap<>(); Map<String, Object> metadata = new HashMap<>();
if (payload.get("title") != null) { if (payload.get("title") != null) {
metadata.put("title", payload.get("title")); metadata.put("title", payload.get("title"));
} }
log.info("[ShortNovel SSE] 开始保存小说: userId={}, queryLength={}, textLength={}",
currentUserId, originalQuery.length(), fullText.length());
Map<String, String> saveResult = epicScriptDialogueServiceImpl.saveNovelResult( Map<String, String> saveResult = epicScriptDialogueServiceImpl.saveNovelResult(
currentUserId, currentUserId,
originalQuery, originalQuery,
@@ -179,18 +192,26 @@ public class ShortNovelServiceImpl implements ShortNovelService {
payload.put("scriptId", saveResult.get("scriptId")); payload.put("scriptId", saveResult.get("scriptId"));
payload.put("conversationId", saveResult.get("conversationId")); payload.put("conversationId", saveResult.get("conversationId"));
payload.put("currentVersionMessageId", saveResult.get("currentVersionMessageId")); payload.put("currentVersionMessageId", saveResult.get("currentVersionMessageId"));
log.info("[ShortNovel SSE] 小说保存成功: scriptId={}", saveResult.get("scriptId"));
} else {
log.warn("[ShortNovel SSE] novel_done 事件缺少 full_text");
} }
} else {
log.warn("[ShortNovel SSE] novel_done 事件跳过保存: originalQuery={}, payload={}",
originalQuery, payload);
} }
} }
// 逐事件转发给前端（OkHttp 每读到一个完整 SSE 事件就立即转发） // 逐事件转发给前端（OkHttp 每读到一个完整 SSE 事件就立即转发）
emitter.send(SseEmitter.event().name(type).data(event.toJSONString())); emitter.send(SseEmitter.event().name(type).data(event.toJSONString()));
emitter.send(SseEmitter.event().comment("")); // 触发 flush
} catch (Exception parseEx) { } catch (Exception parseEx) {
log.warn("SSE 事件解析失败: {}", parseEx.getMessage()); log.warn("SSE 事件解析失败: {}", parseEx.getMessage());
} }
} }
} }
log.info("[ShortNovel SSE] 完成读取上游响应");
emitter.complete(); emitter.complete();
} }
} catch (Exception e) { } catch (Exception e) {