通过okhttp调用SSE流式接口，并将消息返回给客户端(源码)

package com.demo.listener; import com.alibaba.fastjson.JSON; import com.demo.params.req.ChatGlmDto; import lombok.Data; import lombok.extern.slf4j.Slf4j; import okhttp3.Response; import okhttp3.sse.EventSource; import okhttp3.sse.EventSourceListener; import javax.servlet.http.HttpServletResponse; import java.util.concurrent.CountDownLatch; @Slf4j @Data public class SSEListener extends EventSourceListener { private CountDownLatch countDownLatch = new CountDownLatch(1); private ChatGlmDto chatGlmDto; private HttpServletResponse rp; private StringBuffer output = new StringBuffer(); public SSEListener(ChatGlmDto chatGlmDto, HttpServletResponse response) { this.chatGlmDto = chatGlmDto; this.rp = response; } /** * {@inheritDoc} * 建立sse连接 */ @Override public void onOpen(final EventSource eventSource, final Response response) { if (rp != null) { rp.setContentType("text/event-stream"); rp.setCharacterEncoding("UTF-8"); rp.setStatus(200); log.info("建立sse连接..." + JSON.toJSONString(chatGlmDto)); } else { log.info("客户端非sse推送" + JSON.toJSONString(chatGlmDto)); } } /** * 事件 * * @param eventSource * @param id * @param type * @param data */ @Override public void onEvent(EventSource eventSource, String id, String type, String data) { try { output.append(data); if ("finish".equals(type)) { log.info("请求结束{} {}", chatGlmDto.getMessageId(), output.toString()); } if ("error".equals(type)) { log.info("{}: {}source {}", chatGlmDto.getMessageId(), data, JSON.toJSONString(chatGlmDto)); } if (rp != null) { if ("\n".equals(data)) { rp.getWriter().write("event:" + type + "\n"); rp.getWriter().write("id:" + chatGlmDto.getMessageId() + "\n"); rp.getWriter().write("data:\n\n"); rp.getWriter().flush(); } else { String[] dataArr = data.split("\\n"); for (int i = 0; i < dataArr.length; i++) { if (i == 0) { rp.getWriter().write("event:" + type + "\n"); rp.getWriter().write("id:" + chatGlmDto.getMessageId() + "\n"); } if (i == dataArr.length - 1) { rp.getWriter().write("data:" + dataArr[i] + "\n\n"); rp.getWriter().flush(); } else { rp.getWriter().write("data:" + dataArr[i] + "\n"); rp.getWriter().flush(); } } } } } catch (Exception e) { log.error("消息错误[" + JSON.toJSONString(chatGlmDto) + "]", e); countDownLatch.countDown(); throw new RuntimeException(e); } } /** * {@inheritDoc} */ @Override public void onClosed(final EventSource eventSource) { log.info("sse连接关闭:{}", chatGlmDto.getMessageId()); log.info("结果输出:{}" + output.toString()); countDownLatch.countDown(); } /** * {@inheritDoc} */ @Override public void onFailure(final EventSource eventSource, final Throwable t, final Response response) { log.error("使用事件源时出现异常... [响应：{}]...", chatGlmDto.getMessageId()); countDownLatch.countDown(); } public CountDownLatch getCountDownLatch() { return this.countDownLatch; } }