（十二）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<SseEmitterUTF8> 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

<template>
  <div id="app">
    <div style="background: white;text-align: center;">
      <h5>SSE消息</h5>
      <p>{{ sseMsg }}</p>
    </div>
  </div>

</template>

<script>

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));

    }
  },
}
</script>
<style>
body {
  margin: 0px;
  padding: 0px;
}

#app {
  -webkit-background-size: cover;
  -o-background-size: cover;
  background-size: cover;
  margin: 0 auto;
  height: 100%;
}
</style>

⑧查看实现效果

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<ServerSentEvent<Object>> countDown() {
        Flux<ServerSentEvent<Object>> 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 秒";
    }
}

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