springboot实现SSE之牛刀小试

一，概述

1.SSE是何方神圣？

SSE 全称Server Sent Event，直译一下就是服务器发送事件。

其最大的特点，可以简单概括为两个

• 长连接
• 服务端可以向客户端推送信息

2.sse与websocket区别

sse 是单通道，只能服务端向客户端发消息；而 websocket 是双通道

那么为什么有了 websocket 还要搞出一个 sse 呢？既然存在，必然有着它的优越之处

二，实现过程

下面我们以springboot工程为例，实现服务器端不间断向客户端推送数据

1.效果展示

http://124.71.129.204:8080/index

2. 简要流程

3. 源码放送

SSE工具类

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

import org.apache.commons.lang3.RandomStringUtils;
import org.springframework.http.MediaType;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import lombok.extern.slf4j.Slf4j;

/**
 * Server-Sent Events 

 * https://blog.csdn.net/hhl18730252820/article/details/126244274
 */
@Slf4j
public class SSEServer
{
    /**
     * 当前连接数
     */
    private static AtomicInteger count = new AtomicInteger(0);
    
    private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();
    
    public static SseEmitter connect()
    {
        String userId = RandomStringUtils.randomAlphanumeric(10);
        SseEmitter sseEmitter = new SseEmitter(0L); // 设置超时时间，0表示不过期，默认是30秒，超过时间未完成会抛出异常
        
        // 注册回调
        sseEmitter.onCompletion(completionCallBack(userId));
        sseEmitter.onError(errorCallBack(userId));
        sseEmitter.onTimeout(timeOutCallBack(userId));
        sseEmitterMap.put(userId, sseEmitter);
        log.info("create new sse connect ,current user:{}, count: {}", userId, count.incrementAndGet());
        return sseEmitter;
    }
    
    public static void batchSendMessage(String message)
    {
        sseEmitterMap.forEach((k, v) -> {
            try
            {
                v.send(message, MediaType.APPLICATION_JSON);
            }
            catch (IOException e)
            {
                log.error("user id:{}, send message error:{}", k, e.getMessage());
                removeUser(k);
            }
        });
    }
    
    public static void removeUser(String userId)
    {
        sseEmitterMap.remove(userId);
        log.info("remove user id:{}, count: {}", userId, count.decrementAndGet());
    }
    
    private static Runnable completionCallBack(String userId)
    {
        return () -> {
            log.info("结束连接,{}", userId);
            removeUser(userId);
        };
    }
    
    private static Runnable timeOutCallBack(String userId)
    {
        return () -> {
            log.info("连接超时,{}", userId);
            removeUser(userId);
        };
    }
    
    private static Consumer<Throwable> errorCallBack(String userId)
    {
        return throwable -> {
            log.error("连接异常，{}", userId);
            removeUser(userId);
        };
    }
}
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

sse接口

import java.util.concurrent.TimeUnit;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import com.fly.hello.service.SSEServer;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Api(tags = "sse接口")
@RestController
@RequestMapping("/sse")
public class SSEController
{
    long i = -1;
    
    @ApiOperation("初始化")
    @GetMapping("/connect/{userId}")
    public SseEmitter connect(@PathVariable String userId)
    {
        if (i < 0)
        {
            new Thread(() -> sendMessage()).start();
        }
        return SSEServer.connect();
    }
    
    private void sendMessage()
    {
        if (i < 0) // 保证仅触发一次
        {
            log.info("Server-Sent Events start");
            while (true)
            {
                try
                {
                    TimeUnit.MILLISECONDS.sleep(1000);
                }
                catch (InterruptedException e)
                {
                }
                i = ++i % 101;
                SSEServer.batchSendMessage(String.valueOf(i));
            }
        }
    }
}
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

页面引用sse

DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
<html>

<head>
<meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1">
<link href="css/bootstrap.min.css" rel="stylesheet" type="text/css" />
<style>
body {
	margin: 10;
	font-size: 62.5%;
	line-height: 1.5;
}

.blue-button {
	background: #25A6E1;
	padding: 3px 20px;
	color: #fff;
	font-size: 10px;
	border-radius: 2px;
	-moz-border-radius: 2px;
	-webkit-border-radius: 4px;
	border: 1px solid #1A87B9
}

table {
	width: 60%;
}

th {
	background: SteelBlue;
	color: white;
}

td, th {
	border: 1px solid gray;
	font-size: 12px;
	text-align: left;
	padding: 5px 10px;
	overflow: hidden;
	white-space: nowrap;
	text-overflow: ellipsis;
	max-width: 200px;
	white-space: nowrap;
	text-overflow: ellipsis;
	text-overflow: ellipsis;
}
style>
head>
<title>Hello World!title>
<script>
	let data = new EventSource("/sse/connect/001")
	data.onmessage = function(event) {
		document.getElementById("result").innerText = event.data + '%';
		document.getElementById("my-progress").value = event.data;
	}
script>
<body>
	<div class="wrapper-page">
		<table align="center">
			<tr>
				<th colspan="4">Navigateth>
			tr>
			<tr>
				<td><a href="/index" target="_self">indexa>td>
				<td><a href="/404" target="_self">出错页面a>td>
				<td><a href="/doc.html" target="_blank">doc.htmla>td>
				<td><a href="/h2-console" target="_blank">h2-consolea>td>
			tr>
		table>
		<div class="ex-page-content text-center">
			<h2 align="center">
				<a href="index">reloada>
				<div><progress style="width: 60%" id="my-progress" value="0" max="100">progress>div>
				<div id="result">div>
			h2>
			<img src="show/girl" width="600" height="600" />
			<img src="show/pic" width="600" height="600" />
		div>
	div>
body>

html>
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

4.完整项目

https://gitcode.com/00fly/springboot-hello

git clone https://gitcode.com/00fly/springboot-hello.git
1

---

有任何问题和建议，都可以向我提问讨论,大家一起进步，谢谢!

-over-