spring boot 使用SSE向前端推送数据

SSE（Server-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 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";
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

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()
1
2
3
4
5
6
7
8
9
10
11
12
13