• Java中使用Map存储在线用户的集合(登录新增、退出移除)-SpringBoot中集成websocket示例

    场景

    在做服务端与客户端进行通讯时，比如websocket通讯与IM通讯等，需要自定义用户连接/登录时存储用户

    会话信息，断开连接/退出时移除用户会话信息，获取所以当前用户，群发消息等功能。

    所以需要来对用户/客户端信息进行存储。

    若依前后端分离版本地搭建开发环境并运行项目的教程：

    若依前后端分离版手把手教你本地搭建环境并运行项目_霸道流氓气质的博客-CSDN博客_前后端分离项目本地运行

    在上面搭建起来前后端分离的架构基础上，比如集成websocket做一个聊天室功能，需要获取并存取所有的

    连接用户的数据，就需要在连接打开时存储用户，连接关闭时移除用户。

    注：

    博客：
    霸道流氓气质的博客_CSDN博客-C#,架构之路,SpringBoot领域博主
    关注公众号
    霸道的程序猿
    获取编程相关电子书、教程推送与免费下载。

    实现

    1、若依中集成websocket实现获取所有用户

    可参考官方文档-插件集成

    插件集成 | RuoYi

    添加websocket依赖

    1. <!-- SpringBoot Websocket -->
    2. <dependency> 
    3.    <groupId>org.springframework.boot</groupId> 
    4.    <artifactId>spring-boot-starter-websocket</artifactId> 
    5. </dependency>

    其他参考官方说明，这里可以借助其示例代码中存储所有用户信息的实现思路

    新建类WebSocketUsers用来存储所有客户端用户集合

    1. package com.chrisf.websocket;
    2.  
    3. import java.io.IOException;
    4. import java.util.Collection;
    5. import java.util.Map;
    6. import java.util.Set;
    7. import java.util.concurrent.ConcurrentHashMap;
    8. import javax.websocket.Session;
    9. import org.slf4j.Logger;
    10. import org.slf4j.LoggerFactory;
    11.  
    12. /**
    13.  * websocket 客户端用户集
    14.  *
    15.  * @author ruoyi
    16.  */
    17. public class WebSocketUsers
    18. {
    19.     /**
    20.      * WebSocketUsers 日志控制器
    21.      */
    22.     private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketUsers.class);
    23.  
    24.     /**
    25.      * 用户集
    26.      */
    27.     private static Map<String, Session> USERS = new ConcurrentHashMap<String, Session>();
    28.  
    29.     /**
    30.      * 存储用户
    31.      *
    32.      * @param key 唯一键
    33.      * @param session 用户信息
    34.      */
    35.     public static void put(String key, Session session)
    36.     {
    37.         USERS.put(key, session);
    38.     }
    39.  
    40.     /**
    41.      * 移除用户
    42.      *
    43.      * @param session 用户信息
    44.      *
    45.      * @return 移除结果
    46.      */
    47.     public static boolean remove(Session session)
    48.     {
    49.         String key = null;
    50.         boolean flag = USERS.containsValue(session);
    51.         if (flag)
    52.         {
    53.             Set<Map.Entry<String, Session>> entries = USERS.entrySet();
    54.             for (Map.Entry<String, Session> entry : entries)
    55.             {
    56.                 Session value = entry.getValue();
    57.                 if (value.equals(session))
    58.                 {
    59.                     key = entry.getKey();
    60.                     break;
    61.                 }
    62.             }
    63.         }
    64.         else
    65.         {
    66.             return true;
    67.         }
    68.         return remove(key);
    69.     }
    70.  
    71.     /**
    72.      * 移出用户
    73.      *
    74.      * @param key 键
    75.      */
    76.     public static boolean remove(String key)
    77.     {
    78.         LOGGER.info("\n 正在移出用户 - {}", key);
    79.         Session remove = USERS.remove(key);
    80.         if (remove != null)
    81.         {
    82.             boolean containsValue = USERS.containsValue(remove);
    83.             LOGGER.info("\n 移出结果 - {}", containsValue ? "失败" : "成功");
    84.             return containsValue;
    85.         }
    86.         else
    87.         {
    88.             return true;
    89.         }
    90.     }
    91.  
    92.     /**
    93.      * 获取在线用户列表
    94.      *
    95.      * @return 返回用户集合
    96.      */
    97.     public static Map<String, Session> getUsers()
    98.     {
    99.         return USERS;
    100.     }
    101.  
    102.     /**
    103.      * 群发消息文本消息
    104.      *
    105.      * @param message 消息内容
    106.      */
    107.     public static void sendMessageToUsersByText(String message)
    108.     {
    109.         Collection<Session> values = USERS.values();
    110.         for (Session value : values)
    111.         {
    112.             sendMessageToUserByText(value, message);
    113.         }
    114.     }
    115.  
    116.     /**
    117.      * 发送文本消息
    118.      *
    119.      * @param userName 自己的用户名
    120.      * @param message 消息内容
    121.      */
    122.     public static void sendMessageToUserByText(Session session, String message)
    123.     {
    124.         if (session != null)
    125.         {
    126.             try
    127.             {
    128.                 session.getBasicRemote().sendText(message);
    129.             }
    130.             catch (IOException e)
    131.             {
    132.                 LOGGER.error("\n[发送消息异常]", e);
    133.             }
    134.         }
    135.         else
    136.         {
    137.             LOGGER.info("\n[你已离线]");
    138.         }
    139.     }
    140. }

    可以看到这里用的

    private static Map<String, Session> USERS = new ConcurrentHashMap<String, Session>();

    来存取所有的用户会话数据。

    然后在连接建立成功的回调中添加用户

    WebSocketUsers.put(session.getId(), session);

    在连接关闭和抛出异常时删除用户

    WebSocketUsers.remove(sessionId);

    完整的WebsocketServer代码

    1. package com.chrisf.websocket;
    2.  
    3. import java.util.concurrent.Semaphore;
    4. import javax.websocket.OnClose;
    5. import javax.websocket.OnError;
    6. import javax.websocket.OnMessage;
    7. import javax.websocket.OnOpen;
    8. import javax.websocket.Session;
    9. import javax.websocket.server.ServerEndpoint;
    10. import org.slf4j.Logger;
    11. import org.slf4j.LoggerFactory;
    12. import org.springframework.stereotype.Component;
    13.  
    14. /**
    15.  * websocket 消息处理
    16.  *
    17.  * @author ruoyi
    18.  */
    19. @Component
    20. @ServerEndpoint("/chatroom/websocket/message")
    21. public class WebSocketServer
    22. {
    23.     /**
    24.      * WebSocketServer 日志控制器
    25.      */
    26.     private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class);
    27.  
    28.     /**
    29.      * 默认最多允许同时在线人数100
    30.      */
    31.     public static int socketMaxOnlineCount = 100;
    32.  
    33.     private static Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);
    34.  
    35.     /**
    36.      * 连接建立成功调用的方法
    37.      */
    38.     @OnOpen
    39.     public void onOpen(Session session) throws Exception
    40.     {
    41.         boolean semaphoreFlag = false;
    42.         // 尝试获取信号量
    43.         semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore);
    44.         if (!semaphoreFlag)
    45.         {
    46.             // 未获取到信号量
    47.             LOGGER.error("\n 当前在线人数超过限制数- {}", socketMaxOnlineCount);
    48.             WebSocketUsers.sendMessageToUserByText(session, "当前在线人数超过限制数：" + socketMaxOnlineCount);
    49.             session.close();
    50.         }
    51.         else
    52.         {
    53.             // 添加用户
    54.             WebSocketUsers.put(session.getId(), session);
    55.             LOGGER.info("\n 建立连接 - {}", session);
    56.             LOGGER.info("\n 当前人数 - {}", WebSocketUsers.getUsers().size());
    57.             WebSocketUsers.sendMessageToUserByText(session, "连接成功");
    58.         }
    59.     }
    60.  
    61.     /**
    62.      * 连接关闭时处理
    63.      */
    64.     @OnClose
    65.     public void onClose(Session session)
    66.     {
    67.         LOGGER.info("\n 关闭连接 - {}", session);
    68.         // 移除用户
    69.         WebSocketUsers.remove(session.getId());
    70.         // 获取到信号量则需释放
    71.         SemaphoreUtils.release(socketSemaphore);
    72.     }
    73.  
    74.     /**
    75.      * 抛出异常时处理
    76.      */
    77.     @OnError
    78.     public void onError(Session session, Throwable exception) throws Exception
    79.     {
    80.         if (session.isOpen())
    81.         {
    82.             // 关闭连接
    83.             session.close();
    84.         }
    85.         String sessionId = session.getId();
    86.         LOGGER.info("\n 连接异常 - {}", sessionId);
    87.         LOGGER.info("\n 异常信息 - {}", exception);
    88.         // 移出用户
    89.         WebSocketUsers.remove(sessionId);
    90.         // 获取到信号量则需释放
    91.         SemaphoreUtils.release(socketSemaphore);
    92.     }
    93.  
    94.     /**
    95.      * 服务器接收到客户端消息时调用的方法
    96.      */
    97.     @OnMessage
    98.     public void onMessage(String message, Session session)
    99.     {
    100.         String msg = message.replace("你", "我").replace("吗", "");
    101.         WebSocketUsers.sendMessageToUserByText(session, msg);
    102.     }
    103. }

    其他代码

    WebsocketConfig

    1. package com.chrisf.websocket;
    2.  
    3. import org.springframework.context.annotation.Bean;
    4. import org.springframework.context.annotation.Configuration;
    5. import org.springframework.web.socket.server.standard.ServerEndpointExporter;
    6.  
    7. /**
    8.  * websocket 配置
    9.  *
    10.  * @author ruoyi
    11.  */
    12. @Configuration
    13. public class WebSocketConfig
    14. {
    15.     @Bean
    16.     public ServerEndpointExporter serverEndpointExporter()
    17.     {
    18.         return new ServerEndpointExporter();
    19.     }
    20. }

    SemaphoreUtils

    1. package com.chrisf.websocket;
    2.  
    3. import java.util.concurrent.Semaphore;
    4. import org.slf4j.Logger;
    5. import org.slf4j.LoggerFactory;
    6.  
    7. /**
    8.  * 信号量相关处理
    9.  *
    10.  * @author ruoyi
    11.  */
    12. public class SemaphoreUtils
    13. {
    14.     /**
    15.      * SemaphoreUtils 日志控制器
    16.      */
    17.     private static final Logger LOGGER = LoggerFactory.getLogger(SemaphoreUtils.class);
    18.  
    19.     /**
    20.      * 获取信号量
    21.      *
    22.      * @param semaphore
    23.      * @return
    24.      */
    25.     public static boolean tryAcquire(Semaphore semaphore)
    26.     {
    27.         boolean flag = false;
    28.  
    29.         try
    30.         {
    31.             flag = semaphore.tryAcquire();
    32.         }
    33.         catch (Exception e)
    34.         {
    35.             LOGGER.error("获取信号量异常", e);
    36.         }
    37.  
    38.         return flag;
    39.     }
    40.  
    41.     /**
    42.      * 释放信号量
    43.      *
    44.      * @param semaphore
    45.      */
    46.     public static void release(Semaphore semaphore)
    47.     {
    48.  
    49.         try
    50.         {
    51.             semaphore.release();
    52.         }
    53.         catch (Exception e)
    54.         {
    55.             LOGGER.error("释放信号量异常", e);
    56.         }
    57.     }
    58. }

    2、这里的群发消息直接采用的循环所有的用户发送消息，如果用户量较大并且群发消息的频率较高

    可以修改群发消息的逻辑集成自定义线程池。

    SpringBoot集成websocket通过自定义线程池实现高频率、多用户下的群发消息。

    修改上面的WebSocketUsers类

    1. package com.ruoyi.websocket.websocketConfig;
    2.  
    3. import cn.hutool.core.thread.ExecutorBuilder;
    4. import cn.hutool.core.thread.ThreadFactoryBuilder;
    5. import org.slf4j.Logger;
    6. import org.slf4j.LoggerFactory;
    7.  
    8. import javax.websocket.Session;
    9. import java.io.IOException;
    10. import java.util.Collection;
    11. import java.util.Map;
    12. import java.util.Set;
    13. import java.util.concurrent.ConcurrentHashMap;
    14. import java.util.concurrent.ExecutorService;
    15. import java.util.concurrent.LinkedBlockingQueue;
    16.  
    17. /**
    18.  * websocket 客户端用户集
    19.  *
    20.  * @author ruoyi
    21.  */
    22. public class WebSocketUsers {
    23.     /**
    24.      * WebSocketUsers 日志控制器
    25.      */
    26.     private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketUsers.class);
    27.  
    28.     private static ExecutorService pool = ExecutorBuilder.create()
    29.             .setCorePoolSize(WebSocketServer.socketMaxOnlineCount)
    30.             .setMaxPoolSize(WebSocketServer.socketMaxOnlineCount * 2)
    31.             .setWorkQueue(new LinkedBlockingQueue<>(WebSocketServer.socketMaxOnlineCount * 3))
    32.             .setThreadFactory(ThreadFactoryBuilder.create().setNamePrefix("Websocket-Pool-").build())
    33.             .build();
    34.  
    35.     /**
    36.      * 用户集
    37.      */
    38.     private static Map<String, Session> USERS = new ConcurrentHashMap<String, Session>();
    39.  
    40.     /**
    41.      * 存储用户
    42.      *
    43.      * @param key     唯一键
    44.      * @param session 用户信息
    45.      */
    46.     public static void put(String key, Session session) {
    47.         USERS.put(key, session);
    48.     }
    49.  
    50.     /**
    51.      * 移除用户
    52.      *
    53.      * @param session 用户信息
    54.      * @return 移除结果
    55.      */
    56.     public static boolean remove(Session session) {
    57.         String key = null;
    58.         boolean flag = USERS.containsValue(session);
    59.         if (flag) {
    60.             Set<Map.Entry<String, Session>> entries = USERS.entrySet();
    61.             for (Map.Entry<String, Session> entry : entries) {
    62.                 Session value = entry.getValue();
    63.                 if (value.equals(session)) {
    64.                     key = entry.getKey();
    65.                     break;
    66.                 }
    67.             }
    68.         } else {
    69.             return true;
    70.         }
    71.         return remove(key);
    72.     }
    73.  
    74.     /**
    75.      * 移出用户
    76.      *
    77.      * @param key 键
    78.      */
    79.     public static boolean remove(String key) {
    80.         LOGGER.info("\n 正在移出用户 - {}", key);
    81.         Session remove = USERS.remove(key);
    82.         if (remove != null) {
    83.             boolean containsValue = USERS.containsValue(remove);
    84.             LOGGER.info("\n 移出结果 - {}", containsValue ? "失败" : "成功");
    85.             return containsValue;
    86.         } else {
    87.             return true;
    88.         }
    89.     }
    90.  
    91.     /**
    92.      * 获取在线用户列表
    93.      *
    94.      * @return 返回用户集合
    95.      */
    96.     public static Map<String, Session> getUsers() {
    97.         return USERS;
    98.     }
    99.  
    100.     /**
    101.      * 群发消息文本消息
    102.      *
    103.      * @param message 消息内容
    104.      */
    105.     public static void sendMessageToUsersByText(String message) {
    106.         Collection<Session> values = USERS.values();
    107.         for (Session value : values) {
    108.             pool.submit(new Runnable() {
    109.                 @Override
    110.                 public void run() {
    111.                     sendMessageToUserByText(value, message);
    112.                 }
    113.             });
    114.         }
    115.     }
    116.  
    117.     /**
    118.      * 发送文本消息
    119.      *
    120.      * @param userName 自己的用户名
    121.      * @param message  消息内容
    122.      */
    123.     public static void sendMessageToUserByText(Session session, String message) {
    124.         if (session != null) {
    125.             try {
    126.                 synchronized (session) {
    127.                     session.getBasicRemote().sendText(message);
    128.                 }
    129.             } catch (IOException e) {
    130.                 e.printStackTrace();
    131.             }
    132.         } else {
    133.             LOGGER.info("\n[你已离线]");
    134.         }
    135.     }
    136. }

    这里需要用到Hutool的ExecutorBuilder，所以需要引入项目依赖

    1.         <dependency>
    2.             <groupId>cn.hutool</groupId>
    3.             <artifactId>hutool-all</artifactId>
    4.             <version>5.8.3</version>
    5.         </dependency>

    这块具体的说明可以参考

    Java中使用Hutool的ExecutorBuilder实现自定义线程池：

    Java中使用Hutool的ExecutorBuilder实现自定义线程池_霸道流氓气质的博客-CSDN博客

    3、以上是集成websocket协议并存储所有的用户，其他类似的比如集成IM的也可以修改后复用

    若依(基于SpringBoot的权限管理系统)集成MobileIMSDK实现IM服务端的搭建:

    若依(基于SpringBoot的权限管理系统)集成MobileIMSDK实现IM服务端的搭建_霸道流氓气质的博客-CSDN博客_mobileimsdk

    比如在上面的基础上实现在IM中存储所有用户的信息并实现群发功能。

    新建IMUsers存储所有的im用户

    1. package com.chrisf.imextend;
    2.  
    3. import cn.hutool.core.thread.ExecutorBuilder;
    4. import cn.hutool.core.thread.ThreadFactoryBuilder;
    5. import com.chrisf.sdk.protocal.Protocal;
    6. import com.chrisf.sdk.utils.LocalSendHelper;
    7. import io.netty.channel.Channel;
    8. import java.util.Collection;
    9. import java.util.Map;
    10. import java.util.Set;
    11. import java.util.concurrent.ConcurrentHashMap;
    12. import java.util.concurrent.ExecutorService;
    13. import java.util.concurrent.LinkedBlockingQueue;
    14.  
    15. /**
    16.  * im 客户端用户集
    17.  *
    18.  * @author badao
    19.  */
    20. public class ImUsers {
    21.  
    22.     private static ExecutorService pool = ExecutorBuilder.create()
    23.             .setCorePoolSize(20)//初始20个线程
    24.             .setMaxPoolSize(40)//最大40个线程
    25.             .setWorkQueue(new LinkedBlockingQueue<>(60))//有界等待队列，最大等待数是60
    26.             .setThreadFactory(ThreadFactoryBuilder.create().setNamePrefix("IM-Pool-").build())//设置线程前缀
    27.             .build();
    28.  
    29.     /**
    30.      * 用户集
    31.      */
    32.     private static Map<String, Channel> USERS = new ConcurrentHashMap<String, Channel>();
    33.  
    34.     /**
    35.      * 存储用户
    36.      *
    37.      * @param key     唯一键
    38.      * @param session 用户信息
    39.      */
    40.     public static void put(String key, Channel session) {
    41.         USERS.put(key, session);
    42.     }
    43.  
    44.     /**
    45.      * 移除用户
    46.      *
    47.      * @param session 用户信息
    48.      * @return 移除结果
    49.      */
    50.     public static boolean remove(Channel session) {
    51.         String key = null;
    52.         boolean flag = USERS.containsValue(session);
    53.         if (flag) {
    54.             Set<Map.Entry<String, Channel>> entries = USERS.entrySet();
    55.             for (Map.Entry<String, Channel> entry : entries) {
    56.                 Channel value = entry.getValue();
    57.                 if (value.equals(session)) {
    58.                     key = entry.getKey();
    59.                     break;
    60.                 }
    61.             }
    62.         } else {
    63.             return true;
    64.         }
    65.         return remove(key);
    66.     }
    67.  
    68.     /**
    69.      * 移出用户
    70.      *
    71.      * @param key 键
    72.      */
    73.     public static boolean remove(String key) {
    74.         Channel remove = USERS.remove(key);
    75.         if (remove != null) {
    76.             boolean containsValue = USERS.containsValue(remove);
    77.             return containsValue;
    78.         } else {
    79.             return true;
    80.         }
    81.     }
    82.  
    83.     /**
    84.      * 获取在线用户列表
    85.      *
    86.      * @return 返回用户集合
    87.      */
    88.     public static Map<String, Channel> getUsers() {
    89.         return USERS;
    90.     }
    91.  
    92.     /**
    93.      * 群发消息文本消息
    94.      *
    95.      * @param protocal 消息内容
    96.      */
    97.     public static void sendMessageToUsersByText(Protocal protocal) {
    98.         Collection<Channel> values = USERS.values();
    99.         for (Channel value : values) {
    100.             pool.submit(() -> sendMessageToUserByText(value, protocal));
    101.         }
    102.     }
    103.  
    104.     /**
    105.      * 发送消息
    106.      *
    107.      * @param session
    108.      * @param protocal
    109.      */
    110.     public static void sendMessageToUserByText(Channel session, Protocal protocal) {
    111.         if (session != null) {
    112.             synchronized (session) {
    113.                 try {
    114.                     LocalSendHelper.sendData(session,protocal,null);
    115.                 } catch (Exception e) {
    116.                     e.printStackTrace();
    117.                 }
    118.             }
    119.         } else {
    120.  
    121.         }
    122.     }
    123. }

    在登录回调中新增用户

    1.  @Override
    2.  public void onUserLoginSucess(String userId, String extra, Channel session)
    3.  {
    4.   //添加用户
    5.   ImUsers.put(userId,session);
    6.   logger.debug("【IM_回调通知OnUserLoginAction_CallBack】用户："+userId+" 上线了！");
    7.  }

    在登出回调中移除用户

    1.  @Override
    2.  public void onUserLogout(String userId, Object obj, Channel session)
    3.  {
    4.   //移除用户
    5.   ImUsers.remove(userId);
    6.   logger.debug("【DEBUG_回调通知OnUserLogoutAction_CallBack】用户："+userId+" 离线了！");
    7.  }

    在需要群发时群发消息

    1.   if(ImTypeUEnum.CHAT_ROOM.getCode() == typeu)
    2.   {
    3.    ImUsers.sendMessageToUsersByText(p);
    4.   }

  • 相关阅读:
    【C语言】栈和队列的相互实现
    c++ - 第10节 - list类
    java php nodejs python旅游网站设计与开发需求分析Springboot SpringcloudVue汇总一览
    当npm下载库失败时可以用cnpm替代
    stata做莫兰指数，结果不显示是怎么回事，求解答
    百日筑基第五天-关于maven
    利用css var函数让你的组件样式输出规范样式API，可定制性更高；
    机器学习__02__机器学习工程实践
    【毕业设计】基于深度学习的植物识别算法 - cnn opencv python
    pnpm和yarn2 pnp的对比
  • 原文地址：https://blog.csdn.net/BADAO_LIUMANG_QIZHI/article/details/126992039