一、EventSource 的优点

简单易用：EventSource 使用简单，基于标准的 HTTP 协议，无需复杂的握手过程。
自动重连：EventSource 具有内置的重连机制，确保连接中断后自动重新连接。
轻量级：EventSource 使用长轮询机制，消耗的资源相对较少，适合低带宽环境。
跨域支持：EventSource 允许在跨域环境下进行通信，通过适当的响应头授权来自不同域的客户端连接。

二、java 中使用

新建一个SseController

@Tag(name = "SSE消息推送")
@RequestMapping("client/sse")
@RestController("clientSseController")
public class SseController {
    @Autowired
    private SseService sseService;

    @Operation(summary = "客户端连接 SSE")
    @GetMapping(path = "/connect/{clientId}", produces = {MediaType.TEXT_EVENT_STREAM_VALUE})
    public SseEmitter connect(@PathVariable String clientId) throws IOException {
        SseEmitter emitter = sseService.connect(clientId);
        return emitter;
    }

    @Operation(summary = "发送信息测试")
    @GetMapping("/sendMessage")
    public String sendMessage() {
        String result = sseService.sendMessage("1", "hello");
        System.out.println(result);
        return result;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

再建一个实现类SseServiceImpl

@Service
@Slf4j
public class SseServiceImpl implements SseService {
    @Autowired
    private RedisUtil redisUtil;

    // 存储所有的 SSE 连接
    private final ConcurrentHashMap<String, SseEmitter> emitters = new ConcurrentHashMap<>();
    // 创建一个定时任务线程池
    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    private final Long TIMEOUT = 30 * 60 * 1000L; // 设置超时时间 (30 分钟)
    private final Long RECONNECT_TIME = 5 * 1000L; // 设置前端重试时间 (5 秒)
    private static final int MAX_CLIENTS = 500; // 限制最多 500 个连接
    private static final String REDIS_KEY = "sse:"; // redis键值

    /**
     * 连接 SSE
     * @param clientId 用户id
     * @return
     */
    public SseEmitter connect(String clientId) {
        if (emitters.size() >= MAX_CLIENTS) {
            throw new BaseException("SSE 连接数已满");
        }
        SseEmitter emitter = new SseEmitter(TIMEOUT);

        // 注册连接
        emitters.put(clientId, emitter);

        // 在redis记录
        String redisKey = REDIS_KEY + clientId;
        redisUtil.set(redisKey, 0); // 设置一个已经重连的次数

        // 处理连接关闭
        emitter.onTimeout(() -> {
            emitter.complete(); // 关闭 SSE 连接，避免全局异常报错
            removeEmitter(clientId, redisKey);
        });
        emitter.onCompletion(() -> removeEmitter(clientId, redisKey));
        emitter.onError((e) -> removeEmitter(clientId, redisKey));

        return emitter;
    }

    /**
     * 发送消息
     * @param clientId 用户id
     * @param message 消息
     * @return
     */
    public String sendMessage(String clientId, Object message) {
        System.out.println(emitters);
        String redisKey = REDIS_KEY + clientId;
        SseEmitter emitter = emitters.get(clientId);
        if (emitter != null) {
            try {
                emitter.send(SseEmitter.event().reconnectTime(RECONNECT_TIME).data(message)); // 发送消息
                return "消息已发送";
            } catch (IOException e) {
                removeEmitter(clientId, redisKey);
                return "发送失败";
            }
        } else {
            // 可能是 TIMEOUT 正在重连，则 10秒后再次发送(尝试 3 次)
            Object tryNum = redisUtil.get(REDIS_KEY + clientId);
            Integer tryNumInt = tryNum == null ? 0 : (Integer) tryNum;
            if (tryNumInt < 3) {
                // 这里是为了补发数据，防止用户重连期间数据丢失
                scheduler.schedule(() -> {
                    System.out.println("延迟 10 秒后执行：" + tryNumInt);
                    redisUtil.set(redisKey, tryNumInt + 1);
                    sendMessage(clientId, message);
                }, 10, TimeUnit.SECONDS);
            } else {
                removeEmitter(clientId, redisKey);
            }
            return "客户端不存在或连接已关闭";
        }
    }

    private void removeEmitter(String clientId, String redisKey) {
        emitters.remove(clientId);
        redisUtil.del(redisKey);
    }
}

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

需要通知的地方调用:

// 模拟给前端传输json
JSONObject jsonObject = new JSONObject();
            jsonObject.put("promptId", 1);
            jsonObject.put("ossUrl","http://www.baidu.com");
            String result = sseService.sendMessage(1 // 用户id, jsonObject);
            System.out.println(result);

1
2
3
4
5
6

三、js 中使用

let eventSource = null
function connectSSE() {
  if (typeof (EventSource) !== 'undefined') {
    const clientId = 1 // 唯一标识符，这里模拟用户id
    eventSource = new EventSource(`http://localhost:8000/client/sse/connect/${clientId}`)

    eventSource.onopen = function(event) {
      console.log("SSE 连接已建立", event)
    }

    eventSource.onmessage = function(event) {
      console.log("收到消息:", event.data)
      if (event.data) {
        const data = JSON.parse(event.data)
        console.log(data.promptId)
        console.log(data.ossUrl)
      }
    }

    // 监听错误
    eventSource.onerror = function(event) {
      console.log("SSE 连接断开，尝试重连...", event)
    }
  } else {
    console.log('当前浏览器不支持使用EventSource接收服务器推送事件!');
  }
}
connectSSE()

// 关闭 SSE 连接
function closeConnection() {
  eventSource && eventSource.close()
  console.log("SSE 连接已关闭")
}

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

四、注意事项

同一个clientId只会存在一个连接，后面新建的连接会覆盖前面的，前面的连接将不再收到通知。

SseEmitter -- 服务器向前端单向传输数据

发表评论

共0条评论