在springboot中使用Sse(Server-sent Events)Web实时通信技术-服务器发送事件SseEmitter
最近在练习项目时需要用到消息实时推送技术,了解到有两种实时通信技术供我选择:SSE和WebSocket。详细了解后得知SSE是基于http协议,无需导入其他依赖,特点是服务端主动给客户端推送消息(单向),适合浏览器端只做数据接收。而websocket特点是客户端和服务端实现双工通信(双向),多用于即时通信。// 监听窗口关闭事件,主动去关闭sse连接,如果服务端设置永不过期,浏览器关闭后手动清理服
·
最近在练习项目时需要用到消息实时推送技术,了解到有两种实时通信技术供我选择:SSE和WebSocket。详细了解后得知SSE是基于http协议,无需导入其他依赖,特点是服务端主动给客户端推送消息(单向),适合浏览器端只做数据接收。而websocket特点是客户端和服务端实现双工通信(双向),多用于即时通信。基于项目特点,我选择了Sse。而且springboot还整合了sse类名SseEmitter,使用简单方便,在网上copy了一个项目,然后进行改造了一下,以下是记录:
工具类:
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
@Slf4j
public class SseEmitterUtil {
/**
* 当前连接数
*/
private static AtomicInteger count = new AtomicInteger(0);
/**
* 使用map对象,便于根据userId来获取对应的SseEmitter,或者放redis里面
*/
private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();
/**
* 创建用户连接并返回 SseEmitter
*
* @param userId 用户ID
* @return SseEmitter
*/
public static SseEmitter connect(String userId) {
// 设置超时时间,0表示不过期。默认30秒,超过时间未完成会抛出异常:AsyncRequestTimeoutException
SseEmitter sseEmitter = new SseEmitter(0L);
// 注册回调
sseEmitter.onCompletion(completionCallBack(userId));
sseEmitter.onError(errorCallBack(userId));
sseEmitter.onTimeout(timeoutCallBack(userId));
sseEmitterMap.put(userId, sseEmitter);
// 数量+1
count.getAndIncrement();
log.info("创建新的sse连接,当前用户:{}", userId);
return sseEmitter;
}
/**
* 给指定用户发送信息
*/
public static void sendMessage(String userId, String message) {
if (sseEmitterMap.containsKey(userId)) {
try {
// sseEmitterMap.get(userId).send(message, MediaType.APPLICATION_JSON);
sseEmitterMap.get(userId).send(message);
} catch (IOException e) {
log.error("用户[{}]推送异常:{}", userId, e.getMessage());
removeUser(userId);
}
}
}
/**
* 群发消息
*/
public static void batchSendMessage(String wsInfo, List<String> ids) {
ids.forEach(userId -> sendMessage(wsInfo, userId));
}
/**
* 群发所有人
*/
public static void batchSendMessage(String wsInfo) {
sseEmitterMap.forEach((k, v) -> {
try {
v.send(wsInfo, MediaType.APPLICATION_JSON);
} catch (IOException e) {
log.error("用户[{}]推送异常:{}", k, e.getMessage());
removeUser(k);
}
});
}
/**
* 移除用户连接
*/
public static void removeUser(String userId) {
sseEmitterMap.remove(userId);
// 数量-1
count.getAndDecrement();
log.info("移除用户:{}", userId);
}
/**
* 获取当前连接信息
*/
public static List<String> getIds() {
return new ArrayList<>(sseEmitterMap.keySet());
}
/**
* 获取当前连接数量
*/
public static int getUserCount() {
return count.intValue();
}
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.info("连接异常:{}", userId);
removeUser(userId);
};
}
}
controller层:
import com.qian.ssedemo.util.R;
import com.qian.ssedemo.util.SseEmitterUtil;
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;
@RestController
@RequestMapping("/sse")
public class SseController {
/**
* 用于创建连接
*/
@GetMapping("/connect/{userId}")
public SseEmitter connect(@PathVariable String userId) {
return SseEmitterUtil.connect(userId);
}
/**
* 推送给所有人
*
* @param message
* @return
*/
@GetMapping("/push/{message}")
public R<String> push(@PathVariable(name = "message") String message) {
//获取连接人数
int userCount = SseEmitterUtil.getUserCount();
//如果无在线人数,返回
if(userCount<1){
return R.error("无人在线!");
}
SseEmitterUtil.batchSendMessage(message);
return R.success("发送成功!");
}
/**
* 发送给单个人
*
* @param message
* @param userid
* @return
*/
@GetMapping("/push_one/{messsage}/{userid}")
public R<String> pushOne(@PathVariable(name = "message") String message, @PathVariable(name = "userid") String userid) {
SseEmitterUtil.sendMessage(userid, message);
return R.success("推送消息给" + userid);
}
/**
* 关闭连接
*/
@GetMapping("/close/{userid}")
public R<String> close(@PathVariable("userid") String userid) {
SseEmitterUtil.removeUser(userid);
return R.success("连接关闭");
}
}
消息发送客户端(HTMl):
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>消息推送</title>
</head>
<body>
<button onclick="closeSse()">关闭连接</button>
<div id="message"></div>
</body>
<script>
let source = null;
// 用时间戳模拟登录用户
const userId = new Date().getTime();
if (window.EventSource) {
// 建立连接
source = new EventSource('/sse/connect/' + userId);
/**
* 连接一旦建立,就会触发open事件
* 另一种写法:source.onopen = function (event) {}
*/
source.addEventListener('open', function (e) {
setMessageInnerHTML("建立连接。。。");
}, false);
/**
* 客户端收到服务器发来的数据
* 另一种写法:source.onmessage = function (event) {}
*/
source.addEventListener('message', function (e) {
setMessageInnerHTML(e.data);
});
/**
* 如果发生通信错误(比如连接中断),就会触发error事件
* 或者:
* 另一种写法:source.onerror = function (event) {}
*/
source.addEventListener('error', function (e) {
if (e.readyState === EventSource.CLOSED) {
setMessageInnerHTML("连接关闭");
} else {
console.log(e);
}
}, false);
} else {
setMessageInnerHTML("你的浏览器不支持SSE");
}
// 监听窗口关闭事件,主动去关闭sse连接,如果服务端设置永不过期,浏览器关闭后手动清理服务端数据
window.onbeforeunload = function () {
closeSse();
};
// 关闭Sse连接
function closeSse() {
source.close();
const httpRequest = new XMLHttpRequest();
httpRequest.open('GET', '/sse/close/' + userId, true);
httpRequest.send();
console.log("close");
}
// 将消息显示在网页上
function setMessageInnerHTML(innerHTML) {
document.getElementById('message').innerHTML += innerHTML + '<br/>';
}
</script>
</html>
消息发送端(HTMl):
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>发送</title>
<script src="https://cdn.jsdelivr.net/npm/vue@2/dist/vue.js"></script>
<script src="https://unpkg.com/axios/dist/axios.min.js"></script>
</head>
<body>
<div id="app">
<input type="text" v-model="mess" @keyup.enter="send()">
<button type="button" @click="send()">发送</button>
</div>
</body>
<script>
var ff=new Vue({
el: "#app",
data() {
return {
mess: ''
}
},
methods:{
send() {
var t=this;
axios.get("/sse/push/"+t.mess).then(
function(response){
console.log(response.data.code)
if(response.data.code===1){
console.log("发送成功!")
}else {
alert(response.data.msg)
}
t.mess="";
},
function(err){
console.log(err)
}
)
},
}
})
</script>
</html>
完结!
更多推荐
已为社区贡献1条内容
所有评论(0)