• SpringBoot实现SSE构建实时数据单向推送

    • SSE 是一种单向通信，只允许服务器向客户端发送数据。客户端无法向服务器发送数据。
    • SSE 建立在 HTTP 协议之上，使用标准 HTTP 请求和响应。
    • SSE 不需要额外的库或协议处理，客户端可以使用浏览器的原生 EventSource API 来接收数据。
    • SSE 支持跨域通信，可以通过 CORS（跨域资源共享）机制进行配置。
    • SSE 在现代浏览器中也得到广泛支持，但与 WebSocket 相比，它的历史要长一些。

    1. import org.springframework.http.MediaType;
    2. import org.springframework.stereotype.Component;
    3. import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
    4.  
    5. import java.io.IOException;
    6. import java.util.ArrayList;
    7. import java.util.HashMap;
    8. import java.util.List;
    9. import java.util.Map;
    10.  
    11. /**
    12.  * 数据管理器用于管理Server-Sent Events (SSE) 的订阅和数据推送。
    13.  * @author xiaobo
    14.  */
    15. @Component
    16. public class DataManager {
    17.  
    18.     private final Map> dataEmitters = new HashMap<>();
    19.  
    20.     /**
    21.      * 订阅特定数据类型的SSE连接。
    22.      *
    23.      * @param dataType 要订阅的数据类型
    24.      * @param emitter  SSE连接
    25.      */
    26.     public void subscribe(String dataType, SseEmitter emitter) {
    27.         dataEmitters.computeIfAbsent(dataType, k -> new ArrayList<>()).add(emitter);
    28.         emitter.onCompletion(() -> removeEmitter(dataType, emitter));
    29.         emitter.onTimeout(() -> removeEmitter(dataType, emitter));
    30.     }
    31.  
    32.     /**
    33.      * 推送特定数据类型的数据给所有已订阅的连接。
    34.      *
    35.      * @param dataType 要推送的数据类型
    36.      * @param data     要推送的数据
    37.      */
    38.     public void pushData(String dataType, String data) {
    39.         List emitters = dataEmitters.getOrDefault(dataType, new ArrayList<>());
    40.         emitters.forEach(emitter -> {
    41.             try {
    42.                 emitter.send(SseEmitter.event().data(data, MediaType.TEXT_PLAIN));
    43.             } catch (IOException e) {
    44.                 removeEmitter(dataType, emitter);
    45.             }
    46.         });
    47.     }
    48.  
    49.     private void removeEmitter(String dataType, SseEmitter emitter) {
    50.         List emitters = dataEmitters.get(dataType);
    51.         if (emitters != null) {
    52.             emitters.remove(emitter);
    53.         }
    54.     }
    55. }
    56.  

    1. import com.todoitbo.baseSpringbootDasmart.sse.DataManager;  
    2. import org.springframework.http.MediaType;  
    3. import org.springframework.http.ResponseEntity;  
    4. import org.springframework.web.bind.annotation.GetMapping;  
    5. import org.springframework.web.bind.annotation.PathVariable;  
    6. import org.springframework.web.bind.annotation.RequestMapping;  
    7. import org.springframework.web.bind.annotation.RestController;  
    8. import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;  
    9.   
    10. import javax.annotation.Resource;  
    11.   
    12. /**  
    13.  * @author xiaobo  
    14.  */
    15. @RestController  
    16. @RequestMapping("/environment")  
    17. public class EnvironmentController {  
    18.   
    19.     @Resource    private DataManager dataManager;  
    20.   
    21.     @GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)  
    22.     public SseEmitter subscribe() {  
    23.         SseEmitter emitter = new SseEmitter();  
    24.         dataManager.subscribe("environment", emitter);  
    25.         return emitter;  
    26.     }  
    27.   
    28.     // 示例：推送环境监测数据给前端  
    29.     @GetMapping("/push/{testText}")  
    30.     public ResponseEntity pushEnvironmentData(@PathVariable String testText) {  
    31.         dataManager.pushData("environment", testText);  
    32.         return ResponseEntity.ok("Data pushed successfully.");  
    33.     }  
    34. }

     如果没有数据产生会出现连接超时问题。

    默认情况下，EventSource对象会自动重连，以保持连接的持久性。

    第一次订阅SSE连接后，即使后端没有数据产生，之后也能接收到数据。这可以通过定期发送心跳数据

    1. @Scheduled(fixedRate = 30000) // 每30秒发送一次心跳数据
    2. public void sendHeartbeat() {
    3.     dataManager.pushData("heartbeat", "Heartbeat data");
    4. }

    1. html>
    2. <html>
    3. <head>
    4.     <title>SSE Data Receivertitle>
    5. head>
    6. <body>
    7.     <h1>Real-time Data Displayh1>
    8.     <div id="data-container">div>
    9.  
    10.     <script>
    11.         const dataContainer = document.getElementById('data-container');
    12.  
    13.         // 创建一个 EventSource 对象，指定 SSE 服务器端点的 URL
    14.         const eventSource = new EventSource('http://127.0.0.1:13024/environment/subscribe'); // 根据你的控制器端点来设置URL
    15.  
    16.         eventSource.onopen = function(event) {
    17.            
    18.         };
    19.  
    20.         // 添加事件处理程序，监听服务器端发送的事件
    21.         eventSource.onmessage = (event) => {
    22.             const data = event.data;
    23.             // 在这里处理从服务器接收到的数据
    24.             // 可以将数据显示在页面上或进行其他操作
    25.             const newDataElement = document.createElement('p');
    26.             newDataElement.textContent = data;
    27.             dataContainer.appendChild(newDataElement);
    28.         };
    29.  
    30.         eventSource.onerror = (error) => {
    31.              // 处理连接错误
    32.              console.error('Error occurred:', error);
    33.             // 重新建立连接
    34.             eventSource.close();
    35.             setTimeout(() => {
    36.             // 重新建立连接
    37.             eventSource = new EventSource('/environment/subscribe');
    38.     }, 1000); // 1秒后重试
    39.         };
    40.     script>
    41. body>
    42. html>
    43.

     精简版后端

    1. import org.springframework.boot.SpringApplication;
    2. import org.springframework.boot.autoconfigure.SpringBootApplication;
    3. import org.springframework.web.bind.annotation.GetMapping;
    4. import org.springframework.web.bind.annotation.RestController;
    5.  
    6. import java.io.IOException;
    7. import java.util.concurrent.CopyOnWriteArrayList;
    8. import java.util.concurrent.atomic.AtomicLong;
    9.  
    10. @RestController
    11. @SpringBootApplication
    12. public class SseApplication {
    13.  
    14.     private final CopyOnWriteArrayList emitters = new CopyOnWriteArrayList<>();
    15.     private final AtomicLong counter = new AtomicLong();
    16.  
    17.     public static void main(String[] args) {
    18.         SpringApplication.run(SseApplication.class, args);
    19.     }
    20.  
    21.     @GetMapping("/sse")
    22.     public SseEmitter handleSse() {
    23.         SseEmitter emitter = new SseEmitter();
    24.         emitters.add(emitter);
    25.  
    26.         emitter.onCompletion(() -> emitters.remove(emitter));
    27.  
    28.         new Thread(() -> {
    29.             try {
    30.                 for (int i = 0; i < 10; i++) {
    31.                     emitter.send(SseEmitter.event()
    32.                             .id(String.valueOf(counter.incrementAndGet()))
    33.                             .name("message")
    34.                             .data("This is message " + i));
    35.                     Thread.sleep(1000);
    36.                 }
    37.                 emitter.complete();
    38.             } catch (IOException | InterruptedException e) {
    39.                 emitter.completeWithError(e);
    40.             }
    41.         }).start();
    42.  
    43.         return emitter;
    44.     }
    45. }

  • 相关阅读:
    Vue图片懒加载
    专业数据分析思路和常用分析方法
    MAC修改python3命令为py
    计算机视觉（CV）技术
    流水线作业模拟程序
    Spring Boot 2020 官方基础68课程第24个 gateway
    Golang Array 数组使用注意事项和细节
    ISPE GAMP5 中文版
    ros自定义action记录
    搭建Prometheus+Grafana框架监控Hyperledger Fabric的运行
  • 原文地址：https://blog.csdn.net/guo__hang/article/details/134457142