（十二）springboot实战——SSE服务推送事件案例实现

前言

SSE（Server-Sent Events，服务器推送事件）是一种基于HTTP协议的服务器推送技术。它允许服务器向客户端发送异步的、无限长的数据流，而无需客户端不断地轮询或发起请求。这种技术可以用来实现实时通信、在线聊天、即时更新等功能。使用SSE时，客户端通过简单的HTTP请求与服务器建立连接，并在连接保持打开的情况下接收服务器发送的数据。服务器可以随时向客户端发送新的数据（以文本格式），并在数据前面添加特定的标识符，以便客户端能够正确地解析数据。

相较于WebSocket的服务推送，SSE更为轻量级。SSE在客户端与服务器之间建立一个持久的HTTP连接，可以保持连接打开，并通过该连接发送异步的、无限长的数据流。而WebSocket则是一种全双工的协议，支持双向通信，客户端和服务器之间可以直接发送消息。SSE使用文本格式传输数据，并在数据前添加特定的标识符，以便客户端能够正确地解析数据；WebSocket则支持文本和二进制格式的数据传输。由于SSE基于HTTP协议，每次请求和响应都需要经过完整的HTTP协议栈，因此可能存在一定的延迟。而WebSocket则可以更快地建立连接，从而实现更实时的通信。SSE的支持性比WebSocket更广泛，因为它基于标准的HTTP协议，并不需要特殊的服务器支持。而WebSocket则需要WebSocket服务器的支持，因此在某些环境下可能存在兼容性问题。

本节内容主要介绍传统的springboot web项目与响应式springboot webflux项目下如何实现一个简单的SSE服务。

正文

springboot web项目实现SSE案例

①创建一个SSE的接口SseController实现服务数据推送

package com.ht.atp.plat.controller;import com.ht.atp.plat.common.SseEmitterUTF8;import org.springframework.http.MediaType;import org.springframework.scheduling.annotation.Scheduled;import org.springframework.stereotype.Controller;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.io.IOException;import java.nio.charset.StandardCharsets;import java.util.ArrayList;import java.util.List;@Controller@RequestMapping("/sse")public class SseController {    private int count_down_sec = 2 * 60 * 60;    private List emitters = new ArrayList();    /**     * 每秒执行一次     *     */    @Scheduled(fixedDelay = 1000)    public void sendMessage() {        // 获取要发送的消息        String message = getCountDownSec();        byte[] bytes = message.getBytes(StandardCharsets.UTF_8);        for (SseEmitter emitter : emitters) {            try {                emitter.send(bytes, MediaType.TEXT_PLAIN);            } catch (Exception e) {                emitter.completeWithError(e);            }        }    }    @GetMapping(path = "/countDown", produces = MediaType.TEXT_EVENT_STREAM_VALUE)    public SseEmitterUTF8 sse() throws IOException {        SseEmitterUTF8 emitter = new SseEmitterUTF8(0L);        emitter.send("连接成功");        emitter.onCompletion(() -> emitters.remove(emitter));        emitter.onTimeout(() -> emitters.remove(emitter));        emitters.add(emitter);        return emitter;    }    private String getCountDownSec() {        if (count_down_sec > 0) {            int h = count_down_sec / (60 * 60);            int m = (count_down_sec % (60 * 60)) / 60;            int s = (count_down_sec % (60 * 60)) % 60;            count_down_sec--;            return "活动倒计时：" + h + " 小时 " + m + " 分钟 " + s + " 秒";        }        return "活动倒计时：0 小时 0 分钟 0 秒";    }}

② SSE服务说明，通过创建一个MediaType.TEXT_EVENT_STREAM_VALUE类型的HTTP请求，向客户端推送服务数据

 ③创建一个SseEmitterUTF8类继承SseEmitter，解决推送中文消息的服务乱码问题

package com.ht.atp.plat.common;import org.springframework.http.HttpHeaders;import org.springframework.http.MediaType;import org.springframework.http.server.ServerHttpResponse;import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.nio.charset.StandardCharsets;public class SseEmitterUTF8 extends SseEmitter {    public SseEmitterUTF8(Long timeout) {        super(timeout);    }    @Override    protected void extendResponse(ServerHttpResponse outputMessage) {        super.extendResponse(outputMessage);        HttpHeaders headers = outputMessage.getHeaders();        headers.setContentType(new MediaType(MediaType.TEXT_EVENT_STREAM, StandardCharsets.UTF_8));    }}

④ 启动项目，通过接口访问该sse服务

⑤创建一个vue项目，安装vue-sse组件消费推送的消息

命令：

npm i vue-sse -S

 

⑥在main.js中引入vue-sse并注册为一个全局插件

⑦ 使用vue-sse创建一个客户端demo

            
SSE消息
      
{{ sseMsg }}
      export default {  name: 'App',  data() {    return {      sseMsg: '',    }  },  created() {    this.getVueSseMsg();  },  methods: {    getVueSseMsg() {      this.$sse.create('http://127.0.0.1:7777/sse/countDown')          .on('message', (msg) => {            console.info('Message:', msg)            // 处理接收到的消息            this.sseMsg = msg;          })          .on('error', (err) => console.error('Failed to parse or lost connection:', err))          .connect()          .catch((err) => console.error('Failed make initial connection:', err));    }  },}body {  margin: 0px;  padding: 0px;}#app {  -webkit-background-size: cover;  -o-background-size: cover;  background-size: cover;  margin: 0 auto;  height: 100%;}

 ⑧查看实现效果

springboot webflux项目实现SSE案例

①创建一个SSE的接口SseController实现服务数据推送

package com.yundi.atp.controller;import io.swagger.v3.oas.annotations.Operation;import io.swagger.v3.oas.annotations.tags.Tag;import org.springframework.http.MediaType;import org.springframework.http.codec.ServerSentEvent;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;import reactor.core.publisher.Flux;import reactor.util.function.Tuples;import java.time.Duration;@Tag(name = "事件流", description = "事件流")@RestController@RequestMapping("/sse")public class SseController {    private int count_down_sec = 3 * 60 * 60;    @Operation(summary = "获取倒计时数据", description = "获取倒计时数据")    @GetMapping(value = "/countDown", produces = MediaType.TEXT_EVENT_STREAM_VALUE)    public Flux countDown() {        Flux countDown = Flux.interval(Duration.ofSeconds(1))                .map(seq -> Tuples.of(seq, getCountDownSec()))                .map(data -> ServerSentEvent.builder()                        .event("countDown")                        .id(Long.toString(data.getT1()))                        .data(data.getT2())                        .build());        return countDown;    }    private String getCountDownSec() {        if (count_down_sec > 0) {            int h = count_down_sec / (60 * 60);            int m = (count_down_sec % (60 * 60)) / 60;            int s = (count_down_sec % (60 * 60)) % 60;            count_down_sec--;            return "活动倒计时：" + h + " 小时 " + m + " 分钟 " + s + " 秒";        }        return "活动倒计时：0 小时 0 分钟 0 秒";    }}

② 启动项目，使用浏览器访问接口

③将vue的客户端消息sse改为webflux项目的地址

④webflux项目的sse消息效果

结语

本节内容到这里就结束了，我们下期见。。。。。。