【stomp 实战】Spring websocket 用户订阅和会话的管理源码分析

通过Spring websocket 用户校验和业务会话绑定我们学会了如何将业务会话绑定到spring websocket会话上。通过这一节，我们来分析一下会话和订阅的实现

用户会话的数据结构

SessionInfo 用户会话

用户会话定义如下：

private static final class SessionInfo {// subscriptionId -> Subscriptionprivate final Map<String, Subscription> subscriptionMap = new ConcurrentHashMap<>();public Collection<Subscription> getSubscriptions() {return this.subscriptionMap.values();}@Nullablepublic Subscription getSubscription(String subscriptionId) {return this.subscriptionMap.get(subscriptionId);}public void addSubscription(Subscription subscription) {this.subscriptionMap.putIfAbsent(subscription.getId(), subscription);}@Nullablepublic Subscription removeSubscription(String subscriptionId) {return this.subscriptionMap.remove(subscriptionId);}}
  • 用户会话中有subscriptionMap。这个表示一个会话中，可以有多个订阅，可以根据subscriptionId找到订阅。

SessionRegistry 用户会话注册

private static final class SessionRegistry {private final ConcurrentMap<String, SessionInfo> sessions = new ConcurrentHashMap<>();@Nullablepublic SessionInfo getSession(String sessionId) {return this.sessions.get(sessionId);}public void forEachSubscription(BiConsumer<String, Subscription> consumer) {this.sessions.forEach((sessionId, info) ->info.getSubscriptions().forEach(subscription -> consumer.accept(sessionId, subscription)));}public void addSubscription(String sessionId, Subscription subscription) {SessionInfo info = this.sessions.computeIfAbsent(sessionId, _sessionId -> new SessionInfo());info.addSubscription(subscription);}@Nullablepublic SessionInfo removeSubscriptions(String sessionId) {return this.sessions.remove(sessionId);}}
  • SessionRegistry 中sessions 表示多个会话。根据sessionId可以找到唯一一个会话SessionInfo

Subscription 用户订阅

	private static final class Subscription {private final String id;private final String destination;private final boolean isPattern;@Nullableprivate final Expression selector;public Subscription(String id, String destination, boolean isPattern, @Nullable Expression selector) {Assert.notNull(id, "Subscription id must not be null");Assert.notNull(destination, "Subscription destination must not be null");this.id = id;this.selector = selector;this.destination = destination;this.isPattern = isPattern;}public String getId() {return this.id;}public String getDestination() {return this.destination;}public boolean isPattern() {return this.isPattern;}@Nullablepublic Expression getSelector() {return this.selector;}@Overridepublic boolean equals(@Nullable Object other) {return (this == other ||(other instanceof Subscription && this.id.equals(((Subscription) other).id)));}@Overridepublic int hashCode() {return this.id.hashCode();}@Overridepublic String toString() {return "subscription(id=" + this.id + ")";}}

SimpUserRegistry 用户注册接口

用户注册的接口如下：

public interface SimpUserRegistry {/**根据用户名，获取到用户信息* Get the user for the given name.* @param userName the name of the user to look up* @return the user, or {@code null} if not connected*/@NullableSimpUser getUser(String userName);/**获取现在所有的注册的用户* Return a snapshot of all connected users.* <p>The returned set is a copy and will not reflect further changes.* @return the connected users, or an empty set if none*/Set<SimpUser> getUsers();/**获取在线用户数量* Return the count of all connected users.* @return the number of connected users* @since 4.3.5*/int getUserCount();/*** Find subscriptions with the given matcher.* @param matcher the matcher to use* @return a set of matching subscriptions, or an empty set if none*/Set<SimpSubscription> findSubscriptions(SimpSubscriptionMatcher matcher);}

SimpUser实际上就是代表着一个用户，我们来看其实现：LocalSimpUser的定义

	private static class LocalSimpUser implements SimpUser {private final String name;private final Principal user;private final Map<String, SimpSession> userSessions = new ConcurrentHashMap<>(1);public LocalSimpUser(String userName, Principal user) {Assert.notNull(userName, "User name must not be null");this.name = userName;this.user = user;}}

userSessions 表示当前一个用户可以对应多个会话。
这个Principal 是啥，还记得我们上一节通过Spring websocket 用户校验和业务会话绑定中，我们是怎么注册用户的吗

    private void connect(Message<?> message, StompHeaderAccessor accessor) {//1通过请求头获取到tokenString token = accessor.getFirstNativeHeader(WsConstants.TOKEN_HEADER);//2如果token为空或者用户id没有解析出来，抛出异常，spring会将此websocket连接关闭if (StringUtils.isEmpty(token)) {throw new MessageDeliveryException("token missing!");}String userId = TokenUtil.parseToken(token);if (StringUtils.isEmpty(userId)) {throw new MessageDeliveryException("userId missing!");}//这个是每个会话都会有的一个sessionIdString simpleSessionId = (String) message.getHeaders().get(SimpMessageHeaderAccessor.SESSION_ID_HEADER);//3创建自己的业务会话session对象UserSession userSession = new UserSession();userSession.setSimpleSessionId(simpleSessionId);userSession.setUserId(userId);userSession.setCreateTime(LocalDateTime.now());//4关联用户的会话。通过msgOperations.convertAndSendToUser(username, "/topic/subNewMsg", msg); 此方法，可以发送给用户消息accessor.setUser(new UserSessionPrincipal(userSession));}

从token中解析出用户的userId，并通过下面的代码，把当前用户和会话绑定起来。一个用户实际上是可以绑定多个会话的。

 accessor.setUser(new UserSessionPrincipal(userSession));

总结一下用户和会话之间的关系，如下图
在这里插入图片描述

订阅过程的源码分析

前端订阅的代码如下

  stompClient.subscribe("/user/topic/answer", function (response) {createElement("answer", response.body);});

当后端收到订阅消息后，会由SimpleBrokerMessageHandler来处理

	@Overrideprotected void handleMessageInternal(Message<?> message) {MessageHeaders headers = message.getHeaders();String destination = SimpMessageHeaderAccessor.getDestination(headers);String sessionId = SimpMessageHeaderAccessor.getSessionId(headers);updateSessionReadTime(sessionId);if (!checkDestinationPrefix(destination)) {return;}SimpMessageType messageType = SimpMessageHeaderAccessor.getMessageType(headers);if (SimpMessageType.MESSAGE.equals(messageType)) {logMessage(message);sendMessageToSubscribers(destination, message);}else if (SimpMessageType.CONNECT.equals(messageType)) {logMessage(message);if (sessionId != null) {if (this.sessions.get(sessionId) != null) {if (logger.isWarnEnabled()) {logger.warn("Ignoring CONNECT in session " + sessionId + ". Already connected.");}return;}long[] heartbeatIn = SimpMessageHeaderAccessor.getHeartbeat(headers);long[] heartbeatOut = getHeartbeatValue();Principal user = SimpMessageHeaderAccessor.getUser(headers);MessageChannel outChannel = getClientOutboundChannelForSession(sessionId);this.sessions.put(sessionId, new SessionInfo(sessionId, user, outChannel, heartbeatIn, heartbeatOut));SimpMessageHeaderAccessor connectAck = SimpMessageHeaderAccessor.create(SimpMessageType.CONNECT_ACK);initHeaders(connectAck);connectAck.setSessionId(sessionId);if (user != null) {connectAck.setUser(user);}connectAck.setHeader(SimpMessageHeaderAccessor.CONNECT_MESSAGE_HEADER, message);connectAck.setHeader(SimpMessageHeaderAccessor.HEART_BEAT_HEADER, heartbeatOut);Message<byte[]> messageOut = MessageBuilder.createMessage(EMPTY_PAYLOAD, connectAck.getMessageHeaders());getClientOutboundChannel().send(messageOut);}}else if (SimpMessageType.DISCONNECT.equals(messageType)) {logMessage(message);if (sessionId != null) {Principal user = SimpMessageHeaderAccessor.getUser(headers);handleDisconnect(sessionId, user, message);}}else if (SimpMessageType.SUBSCRIBE.equals(messageType)) {logMessage(message);this.subscriptionRegistry.registerSubscription(message);}else if (SimpMessageType.UNSUBSCRIBE.equals(messageType)) {logMessage(message);this.subscriptionRegistry.unregisterSubscription(message);}}

当消息类型为SUBSCRIBE时，会调用subscriptionRegistry.registerSubscription(message)
接着来看下subscriptionRegistry.registerSubscription(message)

//AbstractSubscriptionRegistry@Overridepublic final void registerSubscription(Message<?> message) {MessageHeaders headers = message.getHeaders();SimpMessageType messageType = SimpMessageHeaderAccessor.getMessageType(headers);if (!SimpMessageType.SUBSCRIBE.equals(messageType)) {throw new IllegalArgumentException("Expected SUBSCRIBE: " + message);}String sessionId = SimpMessageHeaderAccessor.getSessionId(headers);if (sessionId == null) {if (logger.isErrorEnabled()) {logger.error("No sessionId in  " + message);}return;}String subscriptionId = SimpMessageHeaderAccessor.getSubscriptionId(headers);if (subscriptionId == null) {if (logger.isErrorEnabled()) {logger.error("No subscriptionId in " + message);}return;}String destination = SimpMessageHeaderAccessor.getDestination(headers);if (destination == null) {if (logger.isErrorEnabled()) {logger.error("No destination in " + message);}return;}addSubscriptionInternal(sessionId, subscriptionId, destination, message);}

这个代码很简单，就是从消息中取出三个东西，sessionId, subscriptionId, destination，进行注册。

//DefaultSubscriptionRegistry@Overrideprotected void addSubscriptionInternal(String sessionId, String subscriptionId, String destination, Message<?> message) {boolean isPattern = this.pathMatcher.isPattern(destination);Expression expression = getSelectorExpression(message.getHeaders());Subscription subscription = new Subscription(subscriptionId, destination, isPattern, expression);this.sessionRegistry.addSubscription(sessionId, subscription);this.destinationCache.updateAfterNewSubscription(sessionId, subscription);}//其实就是添加到sessions map中。会话里把订阅添加到订阅map中public void addSubscription(String sessionId, Subscription subscription) {SessionInfo info = this.sessions.computeIfAbsent(sessionId, _sessionId -> new SessionInfo());info.addSubscription(subscription);}

其实就是添加到sessions map中。会话里把订阅添加到订阅map中

那用户和会话是如何关联起来的？
在这里插入图片描述

  • 当订阅事件发生时，取出当前的Principal( accessor.setUser(xxx)设置的)，然后生成LocalSimpleUser，即用户
  • 把当前会话，添加到当前用户会话中。这样就给用户绑定好了会话了。

用户会话事件

通过Spring事件机制，管理注册用户信息和会话，包括订阅、取消订阅，会话断连。代码如下

//DefaultSimpUserRegistry@Overridepublic void onApplicationEvent(ApplicationEvent event) {AbstractSubProtocolEvent subProtocolEvent = (AbstractSubProtocolEvent) event;Message<?> message = subProtocolEvent.getMessage();MessageHeaders headers = message.getHeaders();String sessionId = SimpMessageHeaderAccessor.getSessionId(headers);Assert.state(sessionId != null, "No session id");if (event instanceof SessionSubscribeEvent) {LocalSimpSession session = this.sessions.get(sessionId);if (session != null) {String id = SimpMessageHeaderAccessor.getSubscriptionId(headers);String destination = SimpMessageHeaderAccessor.getDestination(headers);if (id != null && destination != null) {session.addSubscription(id, destination);}}}else if (event instanceof SessionConnectedEvent) {Principal user = subProtocolEvent.getUser();if (user == null) {return;}String name = user.getName();if (user instanceof DestinationUserNameProvider) {name = ((DestinationUserNameProvider) user).getDestinationUserName();}synchronized (this.sessionLock) {LocalSimpUser simpUser = this.users.get(name);if (simpUser == null) {simpUser = new LocalSimpUser(name, user);this.users.put(name, simpUser);}LocalSimpSession session = new LocalSimpSession(sessionId, simpUser);simpUser.addSession(session);this.sessions.put(sessionId, session);}}else if (event instanceof SessionDisconnectEvent) {synchronized (this.sessionLock) {LocalSimpSession session = this.sessions.remove(sessionId);if (session != null) {LocalSimpUser user = session.getUser();user.removeSession(sessionId);if (!user.hasSessions()) {this.users.remove(user.getName());}}}}else if (event instanceof SessionUnsubscribeEvent) {LocalSimpSession session = this.sessions.get(sessionId);if (session != null) {String subscriptionId = SimpMessageHeaderAccessor.getSubscriptionId(headers);if (subscriptionId != null) {session.removeSubscription(subscriptionId);}}}}

优雅停机

当服务器停机时，最好给客户端发送断连消息，而不是让客户端过了一段时间发现连接断开。
Spring websocket是如何来实现优雅停机的？

public class SubProtocolWebSocketHandlerimplements WebSocketHandler, SubProtocolCapable, MessageHandler, SmartLifecycle {@Overridepublic final void stop() {synchronized (this.lifecycleMonitor) {this.running = false;this.clientOutboundChannel.unsubscribe(this);}// Proactively notify all active WebSocket sessionsfor (WebSocketSessionHolder holder : this.sessions.values()) {try {holder.getSession().close(CloseStatus.GOING_AWAY);}catch (Throwable ex) {if (logger.isWarnEnabled()) {logger.warn("Failed to close '" + holder.getSession() + "': " + ex);}}}}@Overridepublic final void stop(Runnable callback) {synchronized (this.lifecycleMonitor) {stop();callback.run();}}
}

其奥秘就是其实现了SmartLifecycle。这个是Spring的生命周期接口。我们可以通过实现此接口，在相应的生命周期阶段注册回调事件！
上面的代码，通过调用stop接口，给客户端发送了一个断连的消息。即实现了关机时的主动通知断连。

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

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

怎么让电脑耳机和音响都有声音

怎么让电脑耳机和音响都有声音

电脑耳机音响不能同时用没声音怎么办 一般来说&#xff0c;重新开机后问题能够得到解决。右击“我的电脑”---“属性”---“硬件”---“设备管理器”&#xff0c;打开“声音、视频和游戏控制器”有无问题&#xff0c;即看前面有没有出现黄色的“”。 如果您的 电脑 耳机能正常…
阅读更多...
VMware虚拟机中ubuntu使用记录（4）—— 如何在VMware虚拟机中调用本机电脑的摄像头

VMware虚拟机中ubuntu使用记录（4）—— 如何在VMware虚拟机中调用本机电脑的摄像头

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 前言一、虚拟机调用本机摄像头(1) 启动VMware USB 服务(2) 连接本机摄像头(3) 测试摄像头的连接 前言 通过配置虚拟机调用本机摄像头&#xff0c;用户可以在虚拟机…
阅读更多...
redis集群的创建安装与配置，以及维护

redis集群的创建安装与配置，以及维护

一、redis集群配置 port 7000 cluster-enabled yes cluster-config-file nodes.conf cluster-node-timeout 5000 appendonly yes 二、安装ruby&#xff08;ruby可以充当redis客户端的角色&#xff09; yum -y install ruby ruby-devel rubygems rpm-build gem install …
阅读更多...
【SQL】根据条件分组，并根据条件取最大的这一条数据

【SQL】根据条件分组，并根据条件取最大的这一条数据

数据&#xff0c;当字段A相同时&#xff0c;取字段B数值大的这一条数据 ABC123114223234 期望结果 ABC123234 Oracle SELECT A, B, C FROM (SELECT A, B, C,ROW_NUMBER() OVER (PARTITION BY A ORDER BY B DESC) AS rnFROM 表名 ) WHERE rn 1; MySql SELECT t1.A, t1.B,…
阅读更多...
Redis---------实现商品秒杀业务，包括唯一ID，超卖问题，分布式锁

Redis---------实现商品秒杀业务，包括唯一ID，超卖问题，分布式锁

订单ID必须是唯一 唯一ID构成&#xff1a; 代码生成唯一ID&#xff1a; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; import java.time.LocalDateTime; import java.time.ZoneOffset; import java.tim…
阅读更多...
Java内存中的垃圾回收是如何实现的

Java内存中的垃圾回收是如何实现的

Java内存中的垃圾回收是通过Java虚拟机&#xff08;JVM&#xff09;的垃圾收集器&#xff08;Garbage Collector&#xff09;来实现的。垃圾收集器自动管理Java堆内存中的对象&#xff0c;通过识别并回收不再使用的对象来释放内存空间&#xff0c;从而防止内存泄漏和内存溢出等…
阅读更多...
JAVASE相关知识补充

JAVASE相关知识补充

基础知识 类作为成员变量类型&#xff1a; 接口作为成员变量类型&#xff1a; 接口作为方法的参数或者返回值 Object类&#xff1a; java.lang.object 类object是类层次结构的超&#xff08;根&#xff09;类 每个类都是用object作为父类 所有对象包括数组都实现这个类的…
阅读更多...
【论文阅读】Learning Texture Transformer Network for Image Super-Resolution

【论文阅读】Learning Texture Transformer Network for Image Super-Resolution

Learning Texture Transformer Network for Image Super-Resolution 论文地址Abstract1. 简介2.相关工作2.1单图像超分辨率2.2 Reference-based Image Super-Resolution 3. 方法3.1. Texture TransformerLearnable Texture Extractor 可学习的纹理提取器。Relevance Embedding.…
阅读更多...
什么是binutils-arm-linux-gnueabi

什么是binutils-arm-linux-gnueabi

2024年5月3日&#xff0c;周五晚上 binutils-arm-linux-gnueabi 是针对 ARM 架构的 Linux 系统开发的 GNU Binutils 工具链。Binutils 是一组用于汇编、链接和转换目标文件的工具&#xff0c;包括 as (汇编器)、ld (链接器)、objcopy (目标文件转换工具) 等。 binutils-arm-li…
阅读更多...
【Nginx 开发】Nginx高可用集群

【Nginx 开发】Nginx高可用集群

配置集群 高可用集群配置高可用集群 高可用集群 为什么需要配置Nginx集群&#xff1a; 在实际开发过程中Nginx也有可能宕机&#xff0c;如果我们能够将Nginx配置成高可用&#xff0c;当Nginx宕机之后&#xff0c;在集群中我们可以将请求转发到从Nginx服务器当中&#xff0c;由…
阅读更多...
【代码随想录——数组篇】

【代码随想录——数组篇】

代码随想录——数组篇 2. 二分查找3. 移除元素4. 有序数组的平方5. 长度最小的子数组6. 螺旋矩阵II 2. 二分查找 力扣题目链接 前提&#xff1a; 有序数组数组中无重复元素 代码&#xff1a; &#xff08;版本一&#xff09;左闭右闭区间 class Solution {public int sea…
阅读更多...
AcWing 850. Dijkstra求最短路 II

AcWing 850. Dijkstra求最短路 II

Problem: AcWing 850. Dijkstra求最短路 II 文章目录 思路解题方法复杂度Code 思路 这是一个经典的 Dijkstra 算法问题&#xff0c;我们需要找到从点 1 到点 n 的最短路径。Dijkstra 算法是一种贪心算法&#xff0c;它总是选择当前未访问过的节点中距离最短的一个&#xff0c;然…
阅读更多...
什么是Boot Loader？为什么需要Boot Loader？

什么是Boot Loader？为什么需要Boot Loader？

2024年5月4日&#xff0c;周六下午 什么是Boot Loader 引导加载程序&#xff08;Boot Loader&#xff09;是一种软件&#xff0c;位于计算机系统的启动过程中&#xff0c;其主要功能是在系统上电后负责引导加载操作系统内核。Boot Loader通常存储在计算机系统的固化存储设备&a…
阅读更多...
Content-Type详解

Content-Type详解

阅读更多...
代码随想录算法训练营第14天 | 144. 二叉树的前序遍历 | 145. 二叉树的后序遍历 | 94. 二叉树的中序遍历

代码随想录算法训练营第14天 | 144. 二叉树的前序遍历 | 145. 二叉树的后序遍历 | 94. 二叉树的中序遍历

144. 二叉树的前序遍历 解1: 递归 void preorder(struct TreeNode* root, int *array, int *idx) {if (root ! NULL) {array[(*idx)] root->val;preorder(root->left, array, idx);preorder(root->right, array, idx);} }int* preorderTraversal(struct TreeNode* …
阅读更多...
【代码随想录】day49

【代码随想录】day49

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 一、121. 买卖股票的最佳时机二、122.买卖股票的最佳时机II 一、121. 买卖股票的最佳时机 class Solution { public:int maxProfit(vector<int>& prices…
阅读更多...
Qt QImageWriter类介绍

Qt QImageWriter类介绍

1.简介 QImageWriter 用于写入图像文件的类。它提供了将 QImage 对象保存到不同图像格式文件的功能&#xff0c;包括但不限于 PNG、JPEG、BMP 等。QImageWriter 可以将图像写入文件&#xff0c;也可以写入任何 QIODevice&#xff0c;如 QByteArray&#xff0c;这使得它非常灵活…
阅读更多...
pytorch简单神经网络模型训练

pytorch简单神经网络模型训练

目录 一、导入包 二、数据预处理 三、定义神经网络 四、训练模型和测试模型 五、程序入口 一、导入包 import torch import torch.nn as nn import torch.optim as optim # 导入优化器 from torchvision import datasets, transforms # 导入数据集和数据预处理库 from tor…
阅读更多...
python中type,object,class 三者关系

python中type,object,class 三者关系

type,object,class 三者关系 在python中&#xff0c;所有类的创建关系遵循&#xff1a; type -> int -> 1 type -> class -> obj例如&#xff1a; a 1 b "abc" print(type(1)) # <class int> 返回对象的类型 print(type(int)) …
阅读更多...
基于OpenCv的图像金字塔

基于OpenCv的图像金字塔

⚠申明&#xff1a; 未经许可&#xff0c;禁止以任何形式转载&#xff0c;若要引用&#xff0c;请标注链接地址。 全文共计3077字&#xff0c;阅读大概需要3分钟 &#x1f308;更多学习内容&#xff0c; 欢迎&#x1f44f;关注&#x1f440;【文末】我的个人微信公众号&#xf…
阅读更多...
最新文章

Copyright @ 2022~2023