【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.hqwc.cn/news/661187.html

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

相关文章

Linux内核深入学习 - 中断与异常(上)

中断与异常 中断通常被定义为一个事件&#xff1a;让事件改变处理器执行的指令顺序这样的事件&#xff0c;与CPU芯片内外部硬件电路产生的电信号相对应&#xff01; 中断通常分为同步中断与异步中断&#xff1a; 同步中断指的是当指令执行时&#xff0c;由CPU控制单元产生的…

《QT实用小工具·四十九》QT开发的轮播图

1、概述 源码放在文章末尾 该项目实现了界面轮播图的效果&#xff0c;包含如下特点&#xff1a; 左右轮播 鼠标悬浮切换&#xff0c;无需点击 自动定时轮播 自动裁剪和缩放不同尺寸图片 任意添加、插入、删除 单击事件&#xff0c;支持索引和自定义文本 界面美观&#xff0c;圆…

【MyBatis】 MyBatis框架下的高效数据操作:深入理解增删查改(CRUD)

&#x1f493; 博客主页&#xff1a;从零开始的-CodeNinja之路 ⏩ 收录文章&#xff1a;【MyBatis】 MyBatis框架下的高效数据操作&#xff1a;深入理解增删查改&#xff08;CRUD&#xff09; &#x1f389;欢迎大家点赞&#x1f44d;评论&#x1f4dd;收藏⭐文章 目录 My …

树的中心 树形dp

#include<bits/stdc.h> using namespace std; int n; const int N 100005; // 无向边 int ne[N * 2], e[N * 2], idx; int h[N]; int vis[N];int ans 0x7fffffff;void add(int a, int b) {e[idx] b, ne[idx] h[a], h[a] idx; }int dfs(int u) { // 作为根节点vis[u]…

HotSpot VM概述

许多技术人员只把JVM当成黑盒&#xff0c;要想改善Java应用的性能和扩展性无疑是一项艰巨的任务。若要提高Java性能调优的能力&#xff0c;就必须对现代JVM有一定的认知。 HotSpot VM是JDK 1.3版本之后默认的虚拟机&#xff0c;目前是使用最广泛的Java虚拟机。本文主要介绍HotS…

为什么3D模型材质是透明的?---模大狮模型网

在进行3D建模和渲染过程中&#xff0c;正确的材质设置是保证模型外观逼真和渲染效果良好的关键之一。然而&#xff0c;有时您可能会遇到3D模型材质变成透明的情况&#xff0c;这可能会导致意想不到的效果和渲染结果。本文将探讨一些可能导致3D模型材质变成透明的原因&#xff0…

第7篇:创建Nios II工程之控制LED<二>

Q&#xff1a;上一期我们完成了Quartus硬件工程部分&#xff0c;本期我们创建Nios II软件工程这部分。 A&#xff1a;创建完BSP和Nios II Application之后&#xff0c;在source文件main.c中添加LED控制代码&#xff1a;system.h头文件包含了Platform Designer系统中IP的硬件信…

工业互联网通讯协议—欧姆龙(Fins tcp)

一、场景 近期公司要对欧姆龙CP系列设备的数据采集&#xff0c;于是就研究了下欧姆龙的Fins Tcp协议。 二、Fins Tcp 组成字节说明固定头446494E53 FINS对应的ASCII码的十六进制长度4后面剩余指令的长度命令4 握手固定为&#xff1a;00000000 读写固定为&#xff1a;0000000…

vue3 安装-使用之第一篇

首先需要node版本高于V16.14.1 安装 执行 npm create vitelatest 具体选择按照自己实际需要的来 Project name:项目名称 Select a framework:选择用哪种框架 &#xff08;我选择vue&#xff09; Select a variant: 选择用JS还是TS&#xff08;我选择JS&#xff09;找到项目&…

【记录】Python3| 将 PDF 转换成 HTML/XML(✅⭐⭐⭐⭐pdf2htmlEX)

本文将会被汇总至 【记录】Python3&#xff5c;2024年 PDF 转 XML 或 HTML 的第三方库的使用方式、测评过程以及对比结果&#xff08;汇总&#xff09;&#xff0c;更多其他工具请访问该文章查看。 文章目录 pdf2htmlEX 使用体验与评估1 安装指南2 测试代码3 测试结果3.1 转 HT…

AWS最近宣布Amazon Q现已全面上市

每周跟踪AI热点新闻动向和震撼发展 想要探索生成式人工智能的前沿进展吗&#xff1f;订阅我们的简报&#xff0c;深入解析最新的技术突破、实际应用案例和未来的趋势。与全球数同行一同&#xff0c;从行业内部的深度分析和实用指南中受益。不要错过这个机会&#xff0c;成为AI领…

Zynq 7000 系列之启动模式—SD卡启动

SD卡启动允许设备从SD卡&#xff08;Secure Digital Card&#xff09;上读取引导加载程序或操作系统&#xff0c;从而启动系统。SD卡启动具有一些显著的优点&#xff0c;例如方便性、灵活性和可移植性。通过将必要的启动文件存储在SD卡上&#xff0c;用户可以轻松地更换或更新这…