1.概述
消息推送在日常使用中的场景比较多,比如有人点赞了我的博客或者关注了我,这时我就会收到一条推送消息,以此来吸引我点击或者打开应用。消息推送的方式主要分为两种:web消息推送和移动端消息推送。它将所要发送的信息,发送至用户当前访问的网页或者移动设备。本文主要分析在web端进行消息推送的几种方式,实现用户在web端接收推送消息。
2.消息推送几种方式
web消息推送的方式主要分为两种:一种是主动向服务端请求(简单理解为客户端pull消息)、一种是服务端推送(简单理解为服务端push消息)。两种方式各有利弊:主动向服务端请求会按照一定周期不断去请求服务器,如果客户端数量庞大会对服务端造成极大压力,并且数据具有一定延时性;服务端推送实时性较好,但服务端需要存储客户端会话信息,如果客户端数量较多,服务端查询对应会话压力较大。
2.1 Pull消息
Pull消息主要是客户端发起的操作,定时向服务端进行轮询获取消息,轮询可分为短轮询和长轮询。
短轮询:指定时间间隔,由应用浏览器发送http请求,服务器实时返回消息至客户端,浏览器进行展示;短轮询在前端一般通过JS定时器定时发送请求来实现;
长轮询:是对短轮询的一种优化,客户端发起请求,服务器不会立即返回请求结果,而是将请求挂起等待一段时间,如果此时间段内数据变更,立即响应客户端请求,若是一直无变化则等到指定的超时时间响应请求,客户端重新发起长连接;长轮询在nacos、Kafka、RocketMQ队列中使用较多。
2.2 Push消息
服务端向客户端推送,在一定程度上能节约一部分资源,常用的方式有WebSocket、SSE等,还有一些通过中间件RabbitMQ来实现等。本文主要介绍利用SSE方式和WebSocket方式来推送消息,具体如下:
2.2.1 SSE
SSE(Server-sent events)是一种用于实现服务器向客户端实时推送数据的Web技术。与传统的短轮询和长轮询相比,SSE提供了更高效和实时的数据推送机制。SSE基于HTTP协议,允许服务器将数据以事件流(Event Stream)的形式发送给客户端。客户端通过建立持久的HTTP连接,并监听事件流,可以实时接收服务器推送的数据。SSE的主要特点包括:
简单易用:SSE使用基于文本的数据格式,如纯文本、JSON等,使得数据的发送和解析都相对简单;
单向通信:SSE支持服务器向客户端的单向通信,服务器可以主动推送数据给客户端,而客户端只能接收数据;
实时性:SSE建立长时间的连接,使得服务器可以实时地将数据推送给客户端,而无需客户端频繁地发起请求。
SSE的整体实现思路如下,它的原理其实类似于在线视频播放,视频流会连续不断的推送到浏览器,图如下:
其实可以简单地理解为它是一种单向实时通信技术,一旦客户端与服务端建立连接,只能接收服务端信息,不能向服务端发送信息,且拥有自动重连机制,客户端与服务端断开会进行自动重连,websocket断开不能自动重连,这是SSE优于websocket的地方。
Springboot使用SSE功能发送信息代码如下,由于springboot内嵌sse模块,因此不需要引入额外包:
package com.eckey.lab.controller;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
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.ConcurrentHashMap;
/**
* @Author: Marin
* @CreateTime: 2023-10-08 14:29
* @Description: TODO
* @Version: 1.0
*/
@Slf4j
@RestController
@CrossOrigin //此注解是为了解决测试过程中的跨域问题
@RequestMapping("/sse")
public class SSEController {
/**
* 使用Map对象来存放userId和对应的会话
*/
private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();
/**
* @description: 浏览器端注册,将会话信息存入Map,这种方式会导致一个userId只能与服务器建立一个会话,生产环境慎用这种方式
* @author: Marin
* @date: 2023/10/9 16:51
* @param: [userId]
* @return: org.springframework.web.servlet.mvc.method.annotation.SseEmitter
**/
@GetMapping(path = "/subscribe/{userId}", produces = {MediaType.TEXT_EVENT_STREAM_VALUE})
public SseEmitter subscribe(@PathVariable("userId") String userId) throws IOException {
// 设置超时时间,0表示不过期。默认30秒,超过时间未完成会抛出异常:AsyncRequestTimeoutException
SseEmitter sseEmitter = new SseEmitter(30_000L);
// 设置前端的重试时间为15s,如果不加这个发送一下,前端就不会显示连接成功
sseEmitter.send("连接成功");
// 注册回调
sseEmitter.onCompletion(() -> {
log.info("连接结束:{}", userId);
});
sseEmitter.onError((Throwable throwable) -> {
log.error("连接异常:{}", throwable.getMessage());
});
sseEmitter.onTimeout(() -> {
log.warn("连接超时:{}", userId);
});
//以userId为key,如果一个用户多个设备连接,会不准确
sseEmitterMap.put(userId, sseEmitter);
log.info("创建新的sse连接,当前用户:{}", userId);
return sseEmitter;
}
/**
* @description: 向指定用户发送指定信息
* @author: Marin
* @date: 2023/10/9 16:53
* @param: [userId, message]
* @return: void
**/
@GetMapping(path = "/sendMessage")
public void sendMessage(String userId, String message) {
if (sseEmitterMap.containsKey(userId)) {
try {
log.info("向用户:{},发送消息:{}", userId, message);
sseEmitterMap.get(userId).send(message, MediaType.APPLICATION_JSON);
} catch (IOException e) {
log.error("用户[{}]推送异常:{}", userId, e.getMessage());
removeUser(userId);
}
}
}
/**
* @description: 移除用户
* @author: Marin
* @date: 2023/10/9 16:53
* @param: [userId]
* @return: void
**/
private void removeUser(String userId) {
sseEmitterMap.remove(userId);
log.info("移除用户成功:{}", userId);
}
/**
* @description: 删除与指定用户会话
* @author: Marin
* @date: 2023/10/9 16:54
* @param: [userId]
* @return: void
**/
@GetMapping("/close/{userId}")
public void close(@PathVariable("userId") String userId) {
removeUser(userId);
log.info("关闭连接成功:{}", userId);
}
}
前端测试代码如下:
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>SseEmitter</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('http://127.0.0.1:9090/sse/subscribe/' + userId);
setMessageInnerHTML("连接用户=" + userId);
source.addEventListener('open', function(e) {
setMessageInnerHTML("建立连接。。。");
}, false);
source.addEventListener('message', function(e) {
setMessageInnerHTML(e.data);
});
source.addEventListener('error', function(e) {
if (e.readyState === EventSource.CLOSED) {
setMessageInnerHTML("连接关闭");
} else if (e.target.readyState === EventSource.CONNECTING) {
console.log('Connecting...');
}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', 'http://127.0.0.1:9090/sse/close/'+ userId, true);
httpRequest.send();
console.log("close");
}
// 将消息显示在网页上
function setMessageInnerHTML(innerHTML) {
document.getElementById('message').innerHTML += innerHTML + '<br/>';
}
</script>
</html>
打开前端页面,会出现连接信息,如下所示:
调用信息发送接口,跟据用户id发送指定消息,如下:
发送成功后前端接收并显示在页面上,如下:
2.2.2 WebSocket
WebSocket是一种用于实现实时双向通信的Web技术,它使得客户端和服务器之间的数据交换变得更加简单,允许服务端主动向客户端推送数据。在WebSocket API中,浏览器和服务器只需要完成一次握手,两者之间就直接可以创建持久性的连接,并进行双向数据传输。它与SSE在某些方面有所不同。下面是SSE和WebSocket之间的比较:
数据推送方向:SSE是服务器向客户端的单向通信,服务器可以主动推送数据给客户端。而WebSocket是双向通信,允许服务器和客户端之间进行实时的双向数据交换;
连接建立:SSE使用基于HTTP的长连接,通过普通的HTTP请求和响应来建立连接,从而实现数据的实时推送。WebSocket使用自定义的协议,通过建立WebSocket连接来实现双向通信;
兼容性:由于SSE基于HTTP协议,它可以在大多数现代浏览器中使用,并且不需要额外的协议升级。WebSocket在绝大多数现代浏览器中也得到了支持,但在某些特殊的网络环境下可能会遇到问题;
适用场景:SSE适用于服务器向客户端实时推送数据的场景,如股票价格更新、新闻实时推送等。WebSocket适用于需要实时双向通信的场景,如聊天应用、多人协同编辑等。
WebSocket原理图如下所示:
Springboot整合websocket如下:
1.引入pom
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
2.编写socket配置
package com.eckey.lab.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.web.servlet.ServletContextInitializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
import javax.servlet.ServletContext;
import javax.servlet.ServletException;
/**
* @Author: Marin
* @CreateTime: 2023-10-08 16:26
* @Description: TODO
* @Version: 1.0
*/
@Slf4j
@Configuration
public class WebSocketConfig implements ServletContextInitializer {
/**
* 这个bean的注册,用于扫描带有@ServerEndpoint的注解成为websocket,如果你使用外置的tomcat就不需要该配置文件
*/
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
@Override
public void onStartup(ServletContext servletContext) throws ServletException {
String serverInfo = servletContext.getServerInfo();
log.info("serverInfo:{}", serverInfo);
}
}
3.编写SocketServer代码
package com.eckey.lab.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArraySet;
/**
* @Author: Marin
* @CreateTime: 2023-10-08 15:49
* @Description: TODO
* @Version: 1.0
*/
@Component
@Slf4j
@ServerEndpoint("/websocket/{userId}")
public class WebSocketServer {
//与某个客户端的连接会话,需要通过它来给客户端发送数据
private Session session;
private static final CopyOnWriteArraySet<WebSocketServer> webSockets = new CopyOnWriteArraySet<>();
// 用来存在线连接数
private static final Map<String, Session> sessionPool = new HashMap<String, Session>();
/**
* 链接成功调用的方法
*/
@OnOpen
public void onOpen(Session session, @PathParam(value = "userId") String userId) {
try {
this.session = session;
webSockets.add(this);
sessionPool.put(userId, session);
log.info("websocket消息: 有新的连接,总数为:" + webSockets.size());
} catch (Exception e) {
log.error("");
}
}
/**
* 收到客户端消息后调用的方法
*/
@OnMessage
public void onMessage(String message) {
log.info("websocket消息: 收到客户端消息:" + message);
}
/**
* 此为单点消息
*/
public void sendOneMessage(String userId, String message) {
Session session = sessionPool.get(userId);
if (session != null && session.isOpen()) {
try {
log.info("websocket发送单点消息:" + message);
session.getAsyncRemote().sendText(message);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
4.编写controller代码
package com.eckey.lab.controller;
import com.alibaba.fastjson.JSON;
import com.eckey.lab.config.WebSocketServer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import java.io.IOException;
import java.util.HashMap;
/**
* @Author: Marin
* @CreateTime: 2023-10-08 15:24
* @Description: TODO
* @Version: 1.0
*/
@Slf4j
@RestController
@CrossOrigin
@RequestMapping("/socket")
public class WebSocketController {
@Autowired
private WebSocketServer webSocketServer;
@GetMapping(path = "/publish/{userId}")
public String publish(@PathVariable("userId") String userId, String message) throws IOException {
webSocketServer.sendOneMessage(userId, message);
log.info("信息发送成功!userId:{},message:{}", userId, message);
HashMap maps = new HashMap();
maps.put("code", "0");
maps.put("msg", "success");
return JSON.toJSONString(maps);
}
}
5.测试
客户端发送消息,服务接收到消息,具体如下:
调用服务端接口发送消息,客户端接收到消息,具体如下:
3.小结
1.SSE方式是一种基于TCP协议的单向数据传输方式,当连接建立完成,只能由服务端向客户端发送信息;
2.WebSocket是一种双向通信技术,能够实现客户端和服务端的全双工通信,它在建立连接时使用HTTP协议,其它时候都是直接基于TCP协议进行通信;
3.在选择SSE或者WebSocket时,需要跟据场景、性能损耗进行综合考虑,合理的技术选型能够有效增强服务的健壮性。
4.参考文献
1.https://zhuanlan.zhihu.com/p/634581294
2.https://juejin.cn/post/7122014462181113887
3.https://javaguide.cn/system-design/web-real-time-message-push.html
5.附录
https://gitee.com/Marinc/nacos.git