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

server/2024/9/23 21:04:12/

通过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.ppmy.cn/server/32313.html

相关文章

Python数据分析实验二：Python数据预处理

Python数据分析实验二：Python数据预处理

目录 一、实验目的与要求二、实验任务三、主要程序清单和运行结果&#xff08;一&#xff09;对chipotle.csv文件的销售数据进行分析&#xff08;二&#xff09;对描述泰坦尼克号成员的信息进行可视化和相关分析 四、实验体会 一、实验目的与要求 1、目的&#xff1a;   掌握…
阅读更多...
MySQL45讲(一)(42)

MySQL45讲(一)(42)

该节是有关于数据库权限的文章 grant all privileges on *.* to ua% with grant option; 给与用户ua%最高的权限 revoke all privileges on *.* from ua%; 回收所有权限 注意一点&#xff0c;已经存在的连接是不会有全部权限的的&#xff0c;因为在创建连接的时候&#xff0…
阅读更多...
AcWing 850. Dijkstra求最短路 II

AcWing 850. Dijkstra求最短路 II

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

计算机——磁盘

磁盘介绍 磁盘&#xff08;Disk&#xff09;是计算机存储设备的一种&#xff0c;用于持久存储和读取数据。它以圆盘状的物理结构为基础&#xff0c;通过磁性材料在盘片上制造磁道和磁点&#xff0c;利用磁头来读写数据。 磁盘分类 磁盘的常见类型包括硬盘驱动器&#xff08;…
阅读更多...
vivado Aurora 8B/10B IP核（12）- Setp By Step搭建FPGA工程

vivado Aurora 8B/10B IP核（12）- Setp By Step搭建FPGA工程

Step1:任意创建一个新的空的工程&#xff08;创建工程的具体工程如果还不清楚的看我们教程第一季部分&#xff09;&#xff0c; 并且进入IP CORE列表 右击Customize ip Step2:配置 IP CORE-Core options Step3:配置 IP CORE-GT Selections Step4:配置 IP CORE-Shared Logic 为 …
阅读更多...
极简shell制作

极简shell制作

&#x1f30e;自定义简单shell制作 &#xff08;ps: 文末有完整代码&#xff09; 文章目录&#xff1a; 自定义简单shell制作 简单配置Linux文件 自定义Shell编写 命令行解释器       获取输入的命令       字符串分割       子进程进行进程替换 内建命令…
阅读更多...
Linux---软硬链接

Linux---软硬链接

软链接 我们先学习一下怎样创建软链接文件&#xff0c;指令格式为&#xff1a;ln -s 被链接的文件 生成的链接文件名 我们可以这样记忆&#xff1a;ln是link的简称&#xff0c;s是soft的简称。 我们在下面的图片中就是给test文件生成了一个软链接mytest&#xff1a; 我们来解…
阅读更多...
【 书生·浦语大模型实战营】作业（六）：Lagent AgentLego 智能体应用搭建

【 书生·浦语大模型实战营】作业（六）：Lagent AgentLego 智能体应用搭建

【 书生浦语大模型实战营】作业&#xff08;六&#xff09;&#xff1a;Lagent & AgentLego 智能体应用搭建 &#x1f389;AI学习星球推荐&#xff1a; GoAI的学习社区 知识星球是一个致力于提供《机器学习 | 深度学习 | CV | NLP | 大模型 | 多模态 | AIGC 》各个最新AI方…
阅读更多...
最新文章

Copyright @ 2022~2023