spring boot 使用SSE向前端推送数据-CSDN博客

SSEServer-Sent Events是一种基于HTTP的实时通信协议它允许服务器向客户端发送持久性的数据流。与WebSocket不同的是SSE是单向通信只能由服务器向客户端发送数据。Spring Boot通过Spring WebFlux模块提供了对SSE的支持。下面是一个简单的示例
1、后端

package com.example.springbootmp.controller;

import lombok.SneakyThrows;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.*;

@Controller
@RequestMapping(path = "sse")
@CrossOrigin("*")
public class SseRest {
    private final static Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();

    /**
     * 连接sse服务并向前端推送数据
     * @param id
     * @return
     * @throws IOException
     */
    @GetMapping(path = "subscribe", produces = {MediaType.TEXT_EVENT_STREAM_VALUE})
    @SneakyThrows
    public SseEmitter push(String id) throws IOException {
        // 超时时间设置为5分钟用于演示客户端自动重连
        SseEmitter sseEmitter = new SseEmitter(5_60_000L);
        // 设置前端的重试时间为1s
        sseCache.put(id, sseEmitter);
        ExecutorService executorService= Executors.newFixedThreadPool(1,(Runnable r)->{
            Thread t=new Thread(r);
            t.setDaemon(true);
            return t;
        });
        SseEmitter.SseEventBuilder data = SseEmitter.event().name("message").id(id).data("测试数据");
        executorService.execute(()->{
            while (true){
                if(sseCache.containsKey(id)){
                    System.out.println("发送");
                    try {
                        sseEmitter.send(data);
                        Thread.sleep(2000);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }else {
                    System.out.println("结束");
                    break;
                }
            }
        });



//        while (true){
//            sseEmitter.send("测试数据",MediaType.APPLICATION_JSON);
//            Thread.sleep(1000);
//        }

        // onCompletion(): 结束之后的回调触发
        //sseEmitter.onCompletion(() -> System.out.println("完成"));
        return sseEmitter;
    }

    /**
     * http://127.0.0.1:8080/sse/push?id=7777&content=%E4%BD%A0%E5%93%88aaaaaa
     * @param id
     * @param content
     * @return
     * @throws IOException
     */
    @ResponseBody
    @GetMapping(path = "push")
    public String push(String id, String content) throws IOException {
        SseEmitter sseEmitter = sseCache.get(id);
        if (sseEmitter != null) {
            sseEmitter.send(content);
        }
        return "over";
    }

    @ResponseBody
    @GetMapping(path = "/over/{id}")
    public String over(@PathVariable("id") String id) {
        SseEmitter sseEmitter = sseCache.get(id);
        if (sseEmitter != null) {
            // complete(): 表示执行完毕会断开连接
            sseEmitter.complete();
            sseCache.remove(id);
        }
        return "over";
    }
}

2、前端

var source
//开始建立连接部分
source = new EventSource('http://localhost:9999/sse/subscribe?id=122')
        source.addEventListener(
          'message',
          function(event) {
            console.log('接收数据')
            console.log(event.data)
          },
          false
        )
//关闭连接部分
source.close()