玩一下AI
1. SSE协议
我们都知道tcp，ip，http，https，websocket等等协议，今天了解一个新的协议SSE协议（Server-Sent Events）
SSE（Server-Sent Events） 是一种允许服务器主动向客户端推送数据的轻量级协议，基于 HTTP 长连接，实现 单向通信（服务器→客户端）。它是 W3C 标准，浏览器原生支持，无需额外插件（如 EventSource API）
核心特点与工作原理
- 单向通信：仅服务器向客户端发送数据，适合实时通知、日志流、实时更新等场景。
- 基于 HTTP：客户端通过
GET请求建立连接，服务器返回特殊格式的文本流（text/event-stream），连接保持打开状态，直到服务器主动关闭或超时。 - 自动重连：浏览器内置重连机制，连接断开后自动尝试重新连接。
- 数据格式：每条消息以
\n分隔，支持事件类型、数据内容、重试时间等字段，例如：
data: Hello, SSE! // 数据内容
event: customEvent // 自定义事件类型（可选）
id: 123 // 消息ID（可选）
retry: 5000 // 重连时间（毫秒，可选）
\n
适用于无需双向通信，仅需服务器单向推送数据。【比如现在的 gpt，豆包这个问答形式】
前端客户端可以使用原生的 EventSource API：
// 创建EventSource实例，连接服务器
const eventSource = new EventSource('/sse-endpoint');
// 监听默认事件（"message"）
eventSource.onmessage = (event) => {console.log('Received:', event.data);
};
// 监听自定义事件（如"customEvent"）
eventSource.addEventListener('customEvent', (event) => {console.log('Custom Event:', event.data);
});
// 处理错误
eventSource.onerror = (error) => {console.error('SSE Error:', error);// 浏览器会自动重连，无需手动处理
};
服务端可用的就太多了。（本文以SpringBoot3.4.2为例子）
在知道这个协议之前，我们想要达到gpt这种问答形式，输出内容是一点一点拼接的，该怎么弄呢？是不是还可以用websocket。
| 特性 | SSE | WebSocket |
|---|---|---|
| 通信方向 | 单向（服务器→客户端） | 双向（全双工） |
| 协议 | 基于 HTTP（升级为长连接） | 独立协议（ws:// 或 wss://） |
| 二进制支持 | 仅文本（text/event-stream） | 支持文本和二进制 |
| 自动重连 | 浏览器内置 | 需手动实现 |
| 复杂度 | 简单（服务端实现轻量） | 较复杂（需处理握手、心跳） |
| 适用场景 | 服务器单向推送数据 | 双向交互（聊天、实时协作） |
下面结合Spring Boot 简单用一下SSE
// sse协议测试
@PostMapping(value = "/chat", produces = "text/event-stream;charset=UTF-8")
public SseEmitter streamSseMvc() {SseEmitter emitter = new SseEmitter(30_000L);// 模拟发送消息System.out.println("SSE connection started");ScheduledFuture<?> future = service.scheduleAtFixedRate(() -> {try {String message = "Message at " + System.currentTimeMillis();emitter.send(SseEmitter.event().data(message));} catch (IOException e) {try {emitter.send(SseEmitter.event().name("error").data(Map.of("error", e.getMessage())));} catch (IOException ex) {// ignore}emitter.completeWithError(e);}}, 0, 5, TimeUnit.SECONDS);emitter.onCompletion(() -> {System.out.println("SSE connection completed");});emitter.onTimeout(() -> {System.out.println("SSE connection timed out");emitter.complete();});emitter.onError((e) -> {System.out.println("SSE connection error: " + e.getMessage());emitter.completeWithError(e);});return emitter;
}
在SpringBoot中，用SseEmitter就可以达到这个效果了，它也和Websocket一样有onXXX这种类似的方法。上面是使用一个周期性的任务，来模拟AI生成对话的效果的。emitter.send(SseEmitter.event().data(message)); 这个就是服务端向客户端推送数据。
2. okhttp3+sse+deepseek
简单示例：就问一句话
申请deepseekKey这里就略过了，各位读者自行去申请。【因为deepseek官网示例是用的okhttp，所以我这里也用okhttp了】
我们先准备一个接口
@RestController
@RequestMapping("/deepseek")
public class DeepSeekController {@Resourceprivate DeepSeekUtil deepSeekUtil;/*** 访问deepseek-chat*/@PostMapping(value = "/chat", produces = "text/event-stream;charset=UTF-8")public SseEmitter chatSSE() throws IOException {SseEmitter emitter = new SseEmitter(60000L);deepSeekUtil.sendChatReqStream("123456", "你会MySQL数据库吗？", emitter);return emitter; // 这里把该sse对象返回了}private boolean notModel(String model) {return !"deepseek-chat".equals(model) && !"deepseek-reasoner".equals(model);}
}
可以看到我们创建了一个SseEmitter对象，传给了我们自定义的工具
@Component
public class DeepSeekUtil {public static final String DEEPSEEK_CHAT = "deepseek-chat";public static final String DEEPSEEK_REASONER = "deepseek-reasoner";public static final String url = "https://api.deepseek.com/chat/completions";// 存储每个用户的消息列表private static final ConcurrentHashMap<String, List<Message>> msgList = new ConcurrentHashMap<>();// 1.调用api，然后以以 SSE（server-sent events）的形式以流式发送消息增量。消息流以 data: [DONE] 结尾。public void sendChatReqStream(String uid, String message, SseEmitter sseEmitter) throws IOException {// 1.构建一个普通的聊天body请求AccessRequest tRequest = buildNormalChatRequest(uid, message);OkHttpClient client = new OkHttpClient().newBuilder().build();// 封装请求体参数MediaType mediaType = MediaType.parse("application/json; charset=utf-8");RequestBody body = RequestBody.create(JSON.toJSONString(tRequest), mediaType);// 构建请求和请求头Request request = new Request.Builder().url(url).method("POST", body).addHeader("Content-Type", "application/json").addHeader("Accept", "text/event-stream")// 比如你的key是：s-123456// .addHeader("Authorization", "Bearer s-123456").addHeader("Authorization", "Bearer 你的key").build();// 创建一个监听器SseChatListener listener = new SseChatListener(sseEmitter);RealEventSource eventSource = new RealEventSource(request, listener);eventSource.connect(client);}private AccessRequest buildNormalChatRequest(String uid, String message) {// 这里，我们messages，添加了一条“你会MySQL数据库吗？"，来达到一种对话具有上下文的效果List<Message> messages = msgList.computeIfAbsent(uid, k -> new ArrayList<>());messages.add(new Message("user", message));/*[{"system", "你好, 我是DeepSeek-AI助手！"}, {"user", "你会MySQL数据库吗？"}]*/AccessRequest request = new AccessRequest();request.setMessages(messages);request.setModel(DEEPSEEK_CHAT);request.setResponse_format(Map.of("type", "text"));request.setStream(true); // 设置为truerequest.setTemperature(1.0);request.setTop_p(1.0);return request;}@PostConstructpublic void init() {List<Message> m = new ArrayList<Message>();m.add(new Message("system", "你好, 我是DeepSeek-AI助手！"));// 初始化消息列表msgList.put("123456", m);}
}// 请求体，参考deepseek官网
public class AccessRequest {private List<Message> messages;private String model; // 默认模型为deepseek-chatprivate Double frequency_penalty = 0.0;private Integer max_tokens;private Double presence_penalty = 0.0;//{// "type": "text"//}private Map<String, String> response_format;private Object stop = null; // nullprivate Boolean stream; //如果设置为 True，将会以 SSE（server-sent events）的形式以流式发送消息增量。消息流以 data: [DONE] 结尾。private Object stream_options = null;private Double temperature; // 1private Double top_p; // 1private Object tools; // nullprivate String tool_choice = "none";private Boolean logprobs = false;private Integer top_logprobs = null;// get set
}
监听器
@Slf4j
public class SseChatListener extends EventSourceListener {private SseEmitter sseEmitter;public SseChatListener( SseEmitter sseEmitter) {this.sseEmitter = sseEmitter;}/*** 事件*/@Overridepublic void onEvent(EventSource eventSource, String id, String type, String data) {//log.info("sse数据:{}", data);DeepSeekResponse deepSeekResponse = JSON.parseObject(data, DeepSeekResponse.class);DeepSeekResponse.Choice[] choices = deepSeekResponse.getChoices();try {// 发送给前端【客户端】sseEmitter.send(SseEmitter.event().data(choices[0]));} catch (IOException e) {log.error("数据发送异常");throw new RuntimeException(e);}}/*** 建立sse连接*/@Overridepublic void onOpen(final EventSource eventSource, final Response response) {log.info("建立sse连接... {}");}/*** sse关闭*/@Overridepublic void onClosed(final EventSource eventSource) {log.info("sse连接关闭:{}");}/*** 出错了*/@Overridepublic void onFailure(final EventSource eventSource, final Throwable t, final Response response) {log.error("使用事件源时出现异常......");}
}
// DeepSeekResponse.java
@Data
@NoArgsConstructor
@AllArgsConstructor
public class DeepSeekResponse {private String id;private String object;private Long created;private String model;private String system_fingerprint;private Choice[] choices;@Data@AllArgsConstructor@NoArgsConstructorpublic static class Choice {private Integer index;private Delta delta;private Object logprobs;private String finish_reason;}@Data@AllArgsConstructor@NoArgsConstructorpublic static class Delta {private String content;}
}
然后我们用apifox测试一下：
返回这些信息，然后把ai返回的存起来，具体怎么存，就靠读者自行发挥了，添加到该对话，使该对话具有上下文。【DeepSeek /chat/completions API 是一个“无状态” API，即服务端不记录用户请求的上下文，用户在每次请求时，需将之前所有对话历史拼接好后，传递给对话 API。】
[{"system", "你好, 我是DeepSeek-AI助手！"}, {"user", "你会MySQL数据库吗？"},{"ststem", "是的,我熟悉........"} // 把ai返回的存起来
]
下一次对话的时候，请求体AccessRequest里面的List<Message> messages就向上面那样，再往后添加用户问的消息。
上面的例子还有一些小问题，比如说什么时候断开连接那些的。
3. SpringAI
Spring AI 是一个专注于 AI 工程的应用框架，其目标是将 Spring 生态的 “POJO 构建块” 和模块化设计引入 AI 场景，简化企业数据与第三方模型的对接和使用。
下面快速接入deepseek
<!--maven的pom.xml-->
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>3.4.3</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.feng.ai</groupId><artifactId>spring-ai-chat</artifactId><version>0.0.1-SNAPSHOT</version><name>spring-ai-chat</name><description>spring-ai-chat</description><properties><java.version>21</java.version><spring-ai.version>1.0.0-M6</spring-ai.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId></dependency><!--openAI--><dependency><groupId>org.springframework.ai</groupId><artifactId>spring-ai-openai-spring-boot-starter</artifactId></dependency><dependency><groupId>com.alibaba.fastjson2</groupId><artifactId>fastjson2</artifactId><version>2.0.44</version></dependency></dependencies><repositories><repository><id>spring-snapshots</id><name>Spring Snapshots</name><url>https://repo.spring.io/snapshot</url><releases><enabled>false</enabled></releases><snapshots><enabled>true</enabled></snapshots></repository><repository><name>Central Portal Snapshots</name><id>central-portal-snapshots</id><url>https://central.sonatype.com/repository/maven-snapshots/</url><releases><enabled>false</enabled></releases><snapshots><enabled>true</enabled></snapshots></repository></repositories><dependencyManagement><dependencies><dependency><groupId>org.springframework.ai</groupId><artifactId>spring-ai-bom</artifactId><version>${spring-ai.version}</version><type>pom</type><scope>import</scope></dependency></dependencies></dependencyManagement>
</project>
然后是配置文件
spring:application:name: spring-ai-chatai:# The DeepSeek API doesn't support embeddings, so we need to disable it.openai:embedding:enabled: falsebase-url: https://api.deepseek.comapi-key: 你的keychat:options:model: deepseek-reasoner # 使用推理模型stream-usage: true
controller
@Slf4j
@RestController
@RequestMapping("/sp/deepseek")
public class SpDeepseekController {@Resource( name = "openAiChatModel")private OpenAiChatModel deepseekModel;// 直接回答 --- stream-usage: false//@GetMapping("/simpleChat")//public R chat() {// String call = deepseekModel.call("你好, 你会java吗？");// return R.success().setData("call", call);//}// 流式回答@PostMapping(value = "/streamChat", produces = "text/event-stream;charset=UTF-8")public Flux<SpMessage> streamChat(@RequestBody Map<String, String> p) {String userMessage = p.get("userMessage");String sessionId = p.get("sessionId");Prompt prompt = new Prompt(new UserMessage(userMessage));StringBuilder modelStr = new StringBuilder();return deepseekModel.stream(prompt).doOnSubscribe(subscription -> log.info("SSE 连接已启动: {}", sessionId)).doOnComplete(() -> log.info("SSE 连接已关闭: {}", sessionId)).doOnCancel(() -> log.info("SSE 连接已取消: {}", sessionId)).timeout(Duration.ofSeconds(60)) // 超时设置.filter(chatResponse -> chatResponse.getResult().getOutput().getText() != null) // 过滤掉空的响应.map(chatResponse -> {//log.info("SSE 响应: {}", chatResponse.getResult().getOutput());modelStr.append(chatResponse.getResult().getOutput().getText());return SpMessage.builder().role("system").content(chatResponse.getResult().getOutput().getText()).build();});}
}
**TODO：**上面的对话没有记忆，新的请求来了，ai模型并不会带上以前的场景，故需要记忆化。 记忆化的同时还要注意如果把该会话历史中所有的对话全部传给deepseek的话，可能导致 token 超限，故还需要做一个窗口，避免把太多历史对话传过去了。
4. 延伸-Http远程调用
在不讨论微服务架构模式下，我们平时开发难免会碰到需要远程调用接口的情况，【比如说上面调用deepseek的服务】，那么，我们怎么做才是比较好的方式呢？
一次良好的调用过程，我们应该要考虑这几点：超时处理、重试机制、异常处理、日志记录；
此外，于性能来说，我们要避免频繁创建连接带来的开销，可以使用连接池管理；
① RestTemplate
RestTemplate 是一个同步的 HTTP 客户端，提供了简单的方法来发送 HTTP 请求并处理响应。它支持常见的 HTTP 方法（GET、POST、PUT、DELETE 等），并能自动处理 JSON/XML 的序列化和反序列化，这个也是我们非常熟悉的。
下面由于是基于SpringBoot3.4.3，所以httpclient的版本是httpclient5.
@Configuration
public class RestConfig {@Bean("restTemplate")public RestTemplate restTemplate() {// 使用Apache HttpClient连接池（替代默认的 SimpleClientHttpRequestFactory）PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();connectionManager.setMaxTotal(100); // 最大连接数connectionManager.setDefaultMaxPerRoute(20); // 每个路由的最大连接数CloseableHttpClient httpClient = HttpClients.custom().setConnectionManager(connectionManager).evictIdleConnections(TimeValue.of(10, TimeUnit.SECONDS))// 清理空闲连接.build();HttpComponentsClientHttpRequestFactory factory =new HttpComponentsClientHttpRequestFactory(httpClient);factory.setConnectTimeout(3000); // 连接超时（ms）factory.setReadTimeout(5000); // 读取超时（ms）RestTemplate restTemplate = new RestTemplate(factory);// 添加自定义的错误处理器restTemplate.setErrorHandler(new CustomErrorHandler());// 添加日志拦截器restTemplate.getInterceptors().add(new LoggingInterceptor());return restTemplate;}
}@Slf4j
public class LoggingInterceptor implements ClientHttpRequestInterceptor {@NotNull@Overridepublic ClientHttpResponse intercept(HttpRequest request, @NotNull byte[] body, ClientHttpRequestExecution execution) throws IOException {log.info("请求地址: {} {}", request.getMethod(), request.getURI());log.info("请求头: {}", request.getHeaders());log.info("请求体: {}", new String(body, StandardCharsets.UTF_8));ClientHttpResponse response = execution.execute(request, body);log.info("响应状态码: {}", response.getStatusCode());return response;}
}@Slf4j
public class CustomErrorHandler implements ResponseErrorHandler {@Overridepublic boolean hasError(@NotNull ClientHttpResponse response) throws IOException {// 获取 HTTP 状态码HttpStatusCode statusCode = response.getStatusCode();return statusCode.isError(); // 判断状态码是否为错误状态码 【4xx、5xx是true，执行下面的handleError，其他的就false】}@Overridepublic void handleError(@NotNull URI url, @NotNull HttpMethod method, @NotNull ClientHttpResponse response) throws IOException {log.info("请求地址: {} Method: {}",url, method);HttpStatusCode code = response.getStatusCode();if (code.is4xxClientError()) {log.info("客户端错误：{}", code.value());// xxx} else {log.info("服务器错误：{}", code.value());// xxx}}
}
重试降级机制:
@Configuration
@EnableRetry // 开启重试 -- 需要引入AOP
public class RetryConfig {
}// 在service层调用的时候
@Service
public class OrderService {@Resourceprivate RestTemplate restTemplate;@Retryable(maxAttempts = 3, backoff = @Backoff(delay = 1000, multiplier = 2), // 重试间隔 1s, 2s, 4sretryFor = {Exception.class} // 默认重试所有异常//retryFor = {ResourceAccessException.class} // 仅在网络异常时重试)public String queryOrder(String orderId) {return restTemplate.getForObject("/orders/" + orderId, String.class); // 远程调用}@Recover // 重试全部失败后的降级方法public String fallbackQueryOrder(ResourceAccessException e, String orderId) {return "默认订单";}
}
当然还可以再远程调用那里try catch起来，有异常的时候，throw出去可以被@Retryable捕获。
② RestClient
Spring Framework 6.1 引入了全新的同步 HTTP 客户端 RestClient，它在底层使用了与 RestTemplate 相同的基础设施（比如消息转换器和拦截器），但提供了如同 WebClient 一样的现代、流式（fluent）API，兼顾了简洁性与可复用性。与传统的阻塞式 RestTemplate 相比，RestClient 更加直观易用，同时也保持了对同步调用语境的全量支持
同步调用：RestClient 是一个阻塞式客户端，每次 HTTP 请求都会阻塞调用线程直到响应完成。
流式 API：借鉴 WebClient 的设计风格，所有操作均可链式调用，代码更具可读性和可维护性。
复用基础组件：与 RestTemplate 共用 HTTP 请求工厂、消息转换器、拦截器等组件，便于平滑迁移与统一配置
@Configuration
@Slf4j
public class RestClientConfig {@Bean("serviceARestClient")public RestClient restClientA(@Value("${api-service.a-base-url}") String baseUrl) {// 创建连接池PoolingHttpClientConnectionManager manager = new PoolingHttpClientConnectionManager();manager.setMaxTotal(100);manager.setDefaultMaxPerRoute(20);// 创建HttpClientHttpClient httpClient = HttpClientBuilder.create().setConnectionManager(manager).build();// 创建HttpComponentsClientHttpRequestFactoryHttpComponentsClientHttpRequestFactory factory =new HttpComponentsClientHttpRequestFactory(httpClient);factory.setConnectTimeout(3000);factory.setReadTimeout(5000);return RestClient.builder().baseUrl(baseUrl).defaultHeader(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE).defaultCookie("myCookie", "1234").requestInterceptor(clientRequestInterceptor()).requestFactory(factory) // 连接池与超时.build();}@Bean("serviceBRestClient")public RestClient restClientB(@Value("${api-service.b-base-url}") String baseUrl) {return RestClient.builder().baseUrl(baseUrl).defaultHeader(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE).defaultCookie("myCookie", "1234").requestInterceptor(clientRequestInterceptor()).build();}private ClientHttpRequestInterceptor clientRequestInterceptor() {return (request, body, execution) -> {// 添加统一请求头（如认证信息）request.getHeaders().add("my-head", "head-gggggg");// 日志记录log.debug("Request: {} {}", request.getMethod(), request.getURI());request.getHeaders().forEach((name, values) ->values.forEach(value -> log.debug("Header: {}={}", name, value)));ClientHttpResponse response = execution.execute(request, body);log.debug("Response status: {}", response.getStatusCode());return response;};}
}
简单调用:
@Service
public class AService {@Resource(name = "serviceARestClient")private RestClient restClientA;public String queryA(String a) {return restClientA.get().uri("/api/a?a={a}", a).retrieve().onStatus(HttpStatusCode::is4xxClientError, (request, response) -> {throw new HttpClientErrorException(response.getStatusCode());}).onStatus(HttpStatusCode::is5xxServerError, (request, response) -> {throw new ServerErrorException(response.getStatusCode().toString(), null);}).body(String.class);}// 复杂query参数public String queryA(String a, String b) {return restClientA.get().uri( uriBuilder ->uriBuilder.path("/api/bbb").queryParam("a", 25).queryParam("b", "30").build()).retrieve().onStatus(HttpStatusCode::is4xxClientError, (request, response) -> {throw new HttpClientErrorException(response.getStatusCode());}).onStatus(HttpStatusCode::is5xxServerError, (request, response) -> {throw new ServerErrorException(response.getStatusCode().toString(), null);}).body(String.class);}// postpublic String postA(String a) {HashMap<String, Object> map = new HashMap<>();map.put("a", a); map.put("page", 1); map.put("size", 10);return restClientA.post().uri("/api/post").body(map).retrieve().onStatus(HttpStatusCode::is4xxClientError, (request, response) -> {throw new HttpClientErrorException(response.getStatusCode());}).onStatus(HttpStatusCode::is5xxServerError, (request, response) -> {throw new ServerErrorException(response.getStatusCode().toString(), null);}).body(String.class);}}
③ WebClient
Spring框架中包含的原始web框架Spring web MVC是专门为Servlet API和Servlet容器构建的。响应式堆栈web框架Spring WebFlux是在5.0版本中添加的。它是完全非阻塞的，支持响应式流回压，并运行在诸如Netty、Undertow和Servlet容器之类的服务器上。
这两个web框架都镜像了它们的源模块的名字（Spring-webmvc和Spring-webflux 他们的关系图如下，节选自官网），并在Spring框架中共存。每个模块都是可选的。应用程序可以使用其中一个或另一个模块，或者在某些情况下，两者都使用——例如，Spring MVC控制器与响应式WebClient。它对同步和异步以及流方案都有很好的支持。
非阻塞异步模型：基于 Reactor 库（Mono/Flux）实现异步调用，避免线程阻塞，通过少量线程处理高并发请求，显著提升性能
函数式编程：支持链式调用（Builder 模式）与 Lambda 表达式，代码更简洁
流式传输：支持大文件或实时数据的分块传输（Chunked Data），减少内存占用。
这里就不介绍了。
| 特性 | RestTemplate | RestClient | WebClient |
|---|---|---|---|
| 模型 | 阻塞，同步 | 阻塞，同步，流式 API | 非阻塞，响应式【学习曲线较为陡峭】 |
| API 风格 | 模板方法 (getForObject, exchange 等) | 链式流式 (get().uri()...retrieve()) | 链式流式，支持 Mono/Flux |
| 可扩展性 | 依赖大量重载方法 | 可配置拦截器、初始器，支持自定义消息转换器 | 强大的过滤器、拦截器与背压支持 |
| 性能 | 受限于线程池 | 同 RestTemplate，但更简洁 | 更佳，适合高并发场景 |
| 迁移成本 | 低 | 较低，可自然承接现有 RestTemplate 配置 | 较高，需要重构为响应式编程 |
end. 参考
- https://segmentfault.com/a/1190000021133071 【思否-Spring5的WebClient使用详解】
- https://docs.spring.io/spring-framework/reference/integration/rest-clients.html 【Spring官网】
)
