记一次Spring boot使用stomp协议栈时从服务端发起关闭

本文主要是介绍记一次Spring boot使用stomp协议栈时从服务端发起关闭,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

前言

这篇文章是在开发过程中发生的问题,会主要根据本人在本次解决问题的角度进行分析。面向的是一个即时通信项目,与客户端使用websocket做连接接口,使用spring boot的stomp协议栈进行通信。即如下代码形式:

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {@Overridepublic void registerStompEndpoints(StompEndpointRegistry registry) {registry.addEndpoint("/im/ws").setAllowedOrigins("*").withSockJS();}...
}

在分析过程中发现stomp是一个闭环,很多东西都是私有的,可能人家并不想让使用者从服务端发起关闭吧。但是如果客户端进行了订阅,虽然可以通过拦截器的方法拒绝他的消息,但却无法做到不向该通道进行消息推送,所以需要从服务端主动断掉推送思路有二,一种是通过是通过推送关闭消息让客服端发起关闭,第二种webSocketSession直接直接close进行关闭。

解决思路

我们配置项目支持stomp是通过@EnableWebSocketMessageBroker标签进行配置的,追踪过去发现它引入了DelegatingWebSocketMessageBrokerConfiguration.class。

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
@Documented
@Import(DelegatingWebSocketMessageBrokerConfiguration.class)
public @interface EnableWebSocketMessageBroker {}

继续追踪我们发现配置类WebSocketMessageBrokerConfigurationSupport.class中有个subProtocolWebSocketHandler被声明。

public abstract class WebSocketMessageBrokerConfigurationSupport extends AbstractMessageBrokerConfiguration {...@Beanpublic WebSocketHandler subProtocolWebSocketHandler() {return new SubProtocolWebSocketHandler(clientInboundChannel(), clientOutboundChannel());}...
}

继续分析里面有个函数 handleMessage 是向客户端推送消息的,并且是通过sessionId进行精确推送的。

public class SubProtocolWebSocketHandlerimplements WebSocketHandler, SubProtocolCapable, MessageHandler, SmartLifecycle {..../*** Handle an outbound Spring Message to a WebSocket client.*/@Overridepublic void handleMessage(Message<?> message) throws MessagingException {...}...
}

所以我们可以通过@Resource(name = “subProtocolWebSocketHandler”)进行注入获取SubProtocolWebSocketHandler进行操作。因StompCommand中供服务端使用只有CONNECTED(客户端发起连接时回复连接成功)、RECEIPT(接收消息回复接收成功)、MESSAGE(消息推送)、ERROR(错误)可以使用,并且可以从客户端主动发起的只有MESSAGE和ERROR可以使用,通过ERROR可以知道,推送至客户端后,stomp协议就会被动发起关闭流程。以下是一种演示,当然也可以通过MESSAGE的方式自己写逻辑进行主动关闭。

@Controller
public class WSController {...@Resource(name = "subProtocolWebSocketHandler")private SubProtocolWebSocketHandler subProtocolWebSocketHandler;// 这里只是一个demo程序,我是通过http请求的方式控制关闭具体流程和安全性就要自己掌握了@RequestMapping("/closeAllSession")@ResponseBodypublic String closeAllSession() {// 此处只做演示,具体的用户信息和sessionId的对应可以通过configureClientInboundChannel进行映射String[] array = WebSocketConfig.sessionSet.toArray(new String[0]);WebSocketConfig.sessionSet.clear();StompHeaderAccessor headerAccessor = StompHeaderAccessor.create(StompCommand.ERROR);headerAccessor.setSessionId(array[0]);Message<byte[]> createMessage = MessageBuilder.createMessage(new byte[0], headerAccessor.getMessageHeaders());subProtocolWebSocketHandler.handleMessage(createMessage);return "success";}...
}

至此一个简单的从服务端发起关闭stomp接口也就完成了。

但是。。。

流程上可以看出,发起关闭的是stomp协议分析后发起DISCONNECT消息进行关闭,也就是说如果这个流程被恶意劫持,那么这个连接还是不能被关闭的,安全性上还是欠缺一些,所以我们的第二种方法通过webSocketSession进行关闭。查看了其他相关的方法,比较合适做session处理的还是在 SubProtocolWebSocketHandler 中;我们继续分析发现sessions是private修饰的,并且没有对外的获取map或根据sessionId获取session的方法,所以这里只能pass掉;继续分析在afterConnectionEstablished函数中调用了decorateSession方法中对session进行自定义封装,并且decorateSession是protected修饰的。

public class SubProtocolWebSocketHandlerimplements WebSocketHandler, SubProtocolCapable, MessageHandler, SmartLifecycle {...@Overridepublic void afterConnectionEstablished(WebSocketSession session) throws Exception {// WebSocketHandlerDecorator could close the sessionif (!session.isOpen()) {return;}this.stats.incrementSessionCount(session);session = decorateSession(session);this.sessions.put(session.getId(), new WebSocketSessionHolder(session));findProtocolHandler(session).afterSessionStarted(session, this.clientInboundChannel);}....protected WebSocketSession decorateSession(WebSocketSession session) {return new ConcurrentWebSocketSessionDecorator(session, getSendTimeLimit(), getSendBufferSizeLimit());}...
}

所以我们可以通过继承 SubProtocolWebSocketHandler 重写decorateSession的方式拿到webSocketSession并自己维护。

我们要用自定义的 CustomSubProtocolWebSocketHandler 则需要重写 DelegatingWebSocketMessageBrokerConfiguration 配置类。又因为我们自己实现了 DelegatingWebSocketMessageBrokerConfiguration 所以前面的@EnableWebSocketMessageBroker也就要去掉了。

@Configuration
public class CustomWebSocketMessageBrokerConfiguration extends DelegatingWebSocketMessageBrokerConfiguration {@Beanpublic CustomSubProtocolWebSocketHandler customSubProtocolWebSocketHandler() {return new CustomSubProtocolWebSocketHandler(clientInboundChannel(), clientOutboundChannel());}@Overridepublic WebSocketHandler subProtocolWebSocketHandler() {return customSubProtocolWebSocketHandler();}}

websocket通道处理器 继承 SubProtocolWebSocketHandler 的实现如下:

/*** 自定义的websocket通道处理器*/
public class CustomSubProtocolWebSocketHandler extends SubProtocolWebSocketHandler {protected Map<String, Set<String>> userSessionIdMap = new HashMap<>();protected Map<String, CustomWebSocketSessionDecorator> sessionMap = new HashMap<>();private final Lock mapLock = new ReentrantLock();@Autowiredprivate WsuserServer wsuserServer;public CustomSubProtocolWebSocketHandler(MessageChannel clientInboundChannel,SubscribableChannel clientOutboundChannel) {super(clientInboundChannel, clientOutboundChannel);}@Overrideprotected WebSocketSession decorateSession(WebSocketSession session) {CustomWebSocketSessionDecorator decorator = new CustomWebSocketSessionDecorator(session, getSendTimeLimit(),getSendBufferSizeLimit(), this);sessionMap.put(session.getId(), decorator);return decorator;}/*** 设置用户信息* * @param sessionId* @param userInfo*/public void setUserInfo(String sessionId, WebsocketUserInfo userInfo) {String userUnoinAddress = WebsocketFormat.userUnoinAddress(userInfo);mapLock.lock();try {CustomWebSocketSessionDecorator decorator = sessionMap.get(sessionId);Assert.notNull(decorator, String.format("sessionId [%s] 的decorator为空", sessionId));decorator.setPrincipal(userInfo);Set<String> set = userSessionIdMap.get(userUnoinAddress);if (set == null) {set = new HashSet<>();userSessionIdMap.put(userUnoinAddress, set);}set.add(sessionId);} finally {mapLock.unlock();}}@Overridepublic void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {super.afterConnectionClosed(session, closeStatus);this.sessionClear(session, closeStatus);}/*** 清理session* * @param session* @param closeStatus*/public void sessionClear(WebSocketSession session, CloseStatus closeStatus) {// 关闭websocket sessionmapLock.lock();try {CustomWebSocketSessionDecorator remove = sessionMap.remove(session.getId());if (remove != null) {WebsocketUserInfo userInfo = (WebsocketUserInfo) remove.getPrincipal();if (userInfo != null) {String userUnoinAddress = WebsocketFormat.userUnoinAddress(userInfo);Set<String> set = userSessionIdMap.get(userUnoinAddress);if (set != null) {set.remove(session.getId());if (set.isEmpty()) {userSessionIdMap.remove(userUnoinAddress);}}}}} finally {mapLock.unlock();}}/*** 关闭指定用户* * @param userInfo*/public void closeByUserInfo(WebsocketUserInfo userInfo) {List<WebSocketSession> sessionList = new LinkedList<>();String userUnoinAddress = WebsocketFormat.userUnoinAddress(userInfo);Assert.isTrue(LockActuator.trylock(userUnoinAddress), String.format("锁定[%s]失败", userUnoinAddress));try {WebsocketUserInfoWrapper wrapper = WebsocketUserInfoWrapper.of(userInfo);WebsocketUserInfo cacheUserInfo = wsuserServer.getUserInfo(wrapper);mapLock.lock();try {Set<String> set = userSessionIdMap.get(userUnoinAddress);if (set != null) {Iterator<String> iterator = set.iterator();while (iterator.hasNext()) {String next = iterator.next();CustomWebSocketSessionDecorator customWebSocketSessionDecorator = sessionMap.get(next);if (customWebSocketSessionDecorator == null) {// 此处是将那些无关联的进行清理,防止内存泄漏,从目前逻辑上来看应该不会发生,做个保险iterator.remove();continue;}WebsocketUserInfo tempUserInfo = (WebsocketUserInfo) customWebSocketSessionDecorator.getPrincipal();if (tempUserInfo == null || !tempUserInfo.sameUser(cacheUserInfo)) {// 此处只判断哪些需要关闭,不清理sessionMap和userSessionIdMap,交给close逻辑处理sessionList.add(customWebSocketSessionDecorator);continue;}}if (set.isEmpty()) {userSessionIdMap.remove(userUnoinAddress);}}} finally {mapLock.unlock();}} finally {LockActuator.releaseLock();}for (WebSocketSession session : sessionList) {try {session.close();} catch (Exception e) {// do nothing}}}}

CustomWebSocketSessionDecorator的实现如下:

/*** 自定义的websocket控制器*/
public class CustomWebSocketSessionDecorator extends ConcurrentWebSocketSessionDecorator {protected CustomSubProtocolWebSocketHandler webSocketHandler;protected Principal user;public CustomWebSocketSessionDecorator(WebSocketSession delegate, int sendTimeLimit, int bufferSizeLimit,CustomSubProtocolWebSocketHandler webSocketHandler) {super(delegate, sendTimeLimit, bufferSizeLimit);this.webSocketHandler = webSocketHandler;}public CustomWebSocketSessionDecorator(WebSocketSession delegate, int sendTimeLimit, int bufferSizeLimit,OverflowStrategy overflowStrategy, CustomSubProtocolWebSocketHandler webSocketHandler) {super(delegate, sendTimeLimit, bufferSizeLimit, overflowStrategy);this.webSocketHandler = webSocketHandler;}public void setPrincipal(Principal user) {this.user = user;}@Overridepublic Principal getPrincipal() {return this.user;}@Overridepublic void close() throws IOException {this.close(CloseStatus.GOING_AWAY);}@Overridepublic void close(CloseStatus status) throws IOException {super.close(status);webSocketHandler.sessionClear(this, status);}}

上面CustomWebSocketSessionDecorator的实现中有个setPrincipal,需要注意的是stomp中没有直接将 用户信息 设置到webSocketSession中,而是维护在了 StompSubProtocolHandler.class 中,所以如果我们想让用户信息注入到session中还要主动执行一下保存。

结语

以上是使用SpringBoot封装的stomp消息的情况下从服务端发起关闭连接的两种思路。通过分析来看,stomp更想自己维护session,形成一个闭环,毕竟stomp是一个协议栈,并不只是应用在webSocket上,所以这里的方案也只是适用webSocket的通信方案上。

写了一个相关的demo,可以更好的展示出第二种解决方案:
https://github.com/Pluto-Whong/stomp-demo.git

这篇关于记一次Spring boot使用stomp协议栈时从服务端发起关闭的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!


原文地址:
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.chinasem.cn/article/904003

相关文章

Python 字典 (Dictionary)使用详解

《Python字典(Dictionary)使用详解》字典是python中最重要,最常用的数据结构之一,它提供了高效的键值对存储和查找能力,:本文主要介绍Python字典(Dictionary)... 目录字典1.基本特性2.创建字典3.访问元素4.修改字典5.删除元素6.字典遍历7.字典的高级特性默认字典

Java堆转储文件之1.6G大文件处理完整指南

《Java堆转储文件之1.6G大文件处理完整指南》堆转储文件是优化、分析内存消耗的重要工具,:本文主要介绍Java堆转储文件之1.6G大文件处理的相关资料,文中通过代码介绍的非常详细,需要的朋友可... 目录前言文件为什么这么大?如何处理这个文件?分析文件内容(推荐)删除文件(如果不需要)查看错误来源如何避

SpringBoot整合Dubbo+ZK注册失败的坑及解决

《SpringBoot整合Dubbo+ZK注册失败的坑及解决》使用Dubbo框架时,需在公共pom添加依赖,启动类加@EnableDubbo,实现类用@DubboService替代@Service,配... 目录1.先看下公共的pom(maven创建的pom工程)2.启动类上加@EnableDubbo3.实

SpringBoot整合(ES)ElasticSearch7.8实践

《SpringBoot整合(ES)ElasticSearch7.8实践》本文详细介绍了SpringBoot整合ElasticSearch7.8的教程,涵盖依赖添加、客户端初始化、索引创建与获取、批量插... 目录SpringBoot整合ElasticSearch7.8添加依赖初始化创建SpringBoot项

JAVA覆盖和重写的区别及说明

《JAVA覆盖和重写的区别及说明》非静态方法的覆盖即重写,具有多态性;静态方法无法被覆盖,但可被重写(仅通过类名调用),二者区别在于绑定时机与引用类型关联性... 目录Java覆盖和重写的区别经常听到两种话认真读完上面两份代码JAVA覆盖和重写的区别经常听到两种话1.覆盖=重写。2.静态方法可andro

使用Python构建一个高效的日志处理系统

《使用Python构建一个高效的日志处理系统》这篇文章主要为大家详细讲解了如何使用Python开发一个专业的日志分析工具,能够自动化处理、分析和可视化各类日志文件,大幅提升运维效率,需要的可以了解下... 目录环境准备工具功能概述完整代码实现代码深度解析1. 类设计与初始化2. 日志解析核心逻辑3. 文件处

SpringBoot中六种批量更新Mysql的方式效率对比分析

《SpringBoot中六种批量更新Mysql的方式效率对比分析》文章比较了MySQL大数据量批量更新的多种方法,指出REPLACEINTO和ONDUPLICATEKEY效率最高但存在数据风险,MyB... 目录效率比较测试结构数据库初始化测试数据批量修改方案第一种 for第二种 case when第三种

Java docx4j高效处理Word文档的实战指南

《Javadocx4j高效处理Word文档的实战指南》对于需要在Java应用程序中生成、修改或处理Word文档的开发者来说,docx4j是一个强大而专业的选择,下面我们就来看看docx4j的具体使用... 目录引言一、环境准备与基础配置1.1 Maven依赖配置1.2 初始化测试类二、增强版文档操作示例2.

一文详解如何使用Java获取PDF页面信息

《一文详解如何使用Java获取PDF页面信息》了解PDF页面属性是我们在处理文档、内容提取、打印设置或页面重组等任务时不可或缺的一环,下面我们就来看看如何使用Java语言获取这些信息吧... 目录引言一、安装和引入PDF处理库引入依赖二、获取 PDF 页数三、获取页面尺寸(宽高)四、获取页面旋转角度五、判断

Spring Boot中的路径变量示例详解

《SpringBoot中的路径变量示例详解》SpringBoot中PathVariable通过@PathVariable注解实现URL参数与方法参数绑定,支持多参数接收、类型转换、可选参数、默认值及... 目录一. 基本用法与参数映射1.路径定义2.参数绑定&nhttp://www.chinasem.cnbs