投稿
  1. 极客之音首页
  2. 后端漫谈

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

飞熊 • • 后端漫谈 • 阅读 277

生活中，最使人疲惫的往往不是道路的遥远，而是心中的郁闷;最使人痛苦的往往不是生活的不幸，而是希望的破灭;最使人颓废的往往不是前途的坎坷，而是自信的丧失;最使人绝望的往往不是挫折的打击，而是心灵的死亡。所以我们要有自己的梦想，让梦想的星光指引着我们走出落漠，走出惆怅，带着我们走进自己的理想。

导读：本篇文章讲解 Java中使用Map存储在线用户的集合(登录新增、退出移除)-SpringBoot中集成websocket示例，希望对大家有帮助，欢迎收藏，转发！站点地址：www.bmabk.com，来源：原文

场景

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

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

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

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

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

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

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

注：

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

实现

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

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

插件集成 | RuoYi

添加websocket依赖

<!-- SpringBoot Websocket -->
<dependency> 
   <groupId>org.springframework.boot</groupId> 
   <artifactId>spring-boot-starter-websocket</artifactId> 
</dependency>

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

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

package com.chrisf.websocket;

import java.io.IOException;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import javax.websocket.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * websocket 客户端用户集
 *
 * @author ruoyi
 */
public class WebSocketUsers
{
    /**
     * WebSocketUsers 日志控制器
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketUsers.class);

    /**
     * 用户集
     */
    private static Map<String, Session> USERS = new ConcurrentHashMap<String, Session>();

    /**
     * 存储用户
     *
     * @param key 唯一键
     * @param session 用户信息
     */
    public static void put(String key, Session session)
    {
        USERS.put(key, session);
    }

    /**
     * 移除用户
     *
     * @param session 用户信息
     *
     * @return 移除结果
     */
    public static boolean remove(Session session)
    {
        String key = null;
        boolean flag = USERS.containsValue(session);
        if (flag)
        {
            Set<Map.Entry<String, Session>> entries = USERS.entrySet();
            for (Map.Entry<String, Session> entry : entries)
            {
                Session value = entry.getValue();
                if (value.equals(session))
                {
                    key = entry.getKey();
                    break;
                }
            }
        }
        else
        {
            return true;
        }
        return remove(key);
    }

    /**
     * 移出用户
     *
     * @param key 键
     */
    public static boolean remove(String key)
    {
        LOGGER.info("\n 正在移出用户 - {}", key);
        Session remove = USERS.remove(key);
        if (remove != null)
        {
            boolean containsValue = USERS.containsValue(remove);
            LOGGER.info("\n 移出结果 - {}", containsValue ? "失败" : "成功");
            return containsValue;
        }
        else
        {
            return true;
        }
    }

    /**
     * 获取在线用户列表
     *
     * @return 返回用户集合
     */
    public static Map<String, Session> getUsers()
    {
        return USERS;
    }

    /**
     * 群发消息文本消息
     *
     * @param message 消息内容
     */
    public static void sendMessageToUsersByText(String message)
    {
        Collection<Session> values = USERS.values();
        for (Session value : values)
        {
            sendMessageToUserByText(value, message);
        }
    }

    /**
     * 发送文本消息
     *
     * @param userName 自己的用户名
     * @param message 消息内容
     */
    public static void sendMessageToUserByText(Session session, String message)
    {
        if (session != null)
        {
            try
            {
                session.getBasicRemote().sendText(message);
            }
            catch (IOException e)
            {
                LOGGER.error("\n[发送消息异常]", e);
            }
        }
        else
        {
            LOGGER.info("\n[你已离线]");
        }
    }
}

可以看到这里用的

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

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

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

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

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

WebSocketUsers.remove(sessionId);

完整的WebsocketServer代码

package com.chrisf.websocket;

import java.util.concurrent.Semaphore;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * websocket 消息处理
 *
 * @author ruoyi
 */
@Component
@ServerEndpoint("/chatroom/websocket/message")
public class WebSocketServer
{
    /**
     * WebSocketServer 日志控制器
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class);

    /**
     * 默认最多允许同时在线人数100
     */
    public static int socketMaxOnlineCount = 100;

    private static Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session) throws Exception
    {
        boolean semaphoreFlag = false;
        // 尝试获取信号量
        semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore);
        if (!semaphoreFlag)
        {
            // 未获取到信号量
            LOGGER.error("\n 当前在线人数超过限制数- {}", socketMaxOnlineCount);
            WebSocketUsers.sendMessageToUserByText(session, "当前在线人数超过限制数：" + socketMaxOnlineCount);
            session.close();
        }
        else
        {
            // 添加用户
            WebSocketUsers.put(session.getId(), session);
            LOGGER.info("\n 建立连接 - {}", session);
            LOGGER.info("\n 当前人数 - {}", WebSocketUsers.getUsers().size());
            WebSocketUsers.sendMessageToUserByText(session, "连接成功");
        }
    }

    /**
     * 连接关闭时处理
     */
    @OnClose
    public void onClose(Session session)
    {
        LOGGER.info("\n 关闭连接 - {}", session);
        // 移除用户
        WebSocketUsers.remove(session.getId());
        // 获取到信号量则需释放
        SemaphoreUtils.release(socketSemaphore);
    }

    /**
     * 抛出异常时处理
     */
    @OnError
    public void onError(Session session, Throwable exception) throws Exception
    {
        if (session.isOpen())
        {
            // 关闭连接
            session.close();
        }
        String sessionId = session.getId();
        LOGGER.info("\n 连接异常 - {}", sessionId);
        LOGGER.info("\n 异常信息 - {}", exception);
        // 移出用户
        WebSocketUsers.remove(sessionId);
        // 获取到信号量则需释放
        SemaphoreUtils.release(socketSemaphore);
    }

    /**
     * 服务器接收到客户端消息时调用的方法
     */
    @OnMessage
    public void onMessage(String message, Session session)
    {
        String msg = message.replace("你", "我").replace("吗", "");
        WebSocketUsers.sendMessageToUserByText(session, msg);
    }
}

其他代码

WebsocketConfig

package com.chrisf.websocket;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

/**
 * websocket 配置
 *
 * @author ruoyi
 */
@Configuration
public class WebSocketConfig
{
    @Bean
    public ServerEndpointExporter serverEndpointExporter()
    {
        return new ServerEndpointExporter();
    }
}

SemaphoreUtils

package com.chrisf.websocket;

import java.util.concurrent.Semaphore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 信号量相关处理
 *
 * @author ruoyi
 */
public class SemaphoreUtils
{
    /**
     * SemaphoreUtils 日志控制器
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(SemaphoreUtils.class);

    /**
     * 获取信号量
     *
     * @param semaphore
     * @return
     */
    public static boolean tryAcquire(Semaphore semaphore)
    {
        boolean flag = false;

        try
        {
            flag = semaphore.tryAcquire();
        }
        catch (Exception e)
        {
            LOGGER.error("获取信号量异常", e);
        }

        return flag;
    }

    /**
     * 释放信号量
     *
     * @param semaphore
     */
    public static void release(Semaphore semaphore)
    {

        try
        {
            semaphore.release();
        }
        catch (Exception e)
        {
            LOGGER.error("释放信号量异常", e);
        }
    }
}

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

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

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

修改上面的WebSocketUsers类

package com.ruoyi.websocket.websocketConfig;

import cn.hutool.core.thread.ExecutorBuilder;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.websocket.Session;
import java.io.IOException;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * websocket 客户端用户集
 *
 * @author ruoyi
 */
public class WebSocketUsers {
    /**
     * WebSocketUsers 日志控制器
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketUsers.class);

    private static ExecutorService pool = ExecutorBuilder.create()
            .setCorePoolSize(WebSocketServer.socketMaxOnlineCount)
            .setMaxPoolSize(WebSocketServer.socketMaxOnlineCount * 2)
            .setWorkQueue(new LinkedBlockingQueue<>(WebSocketServer.socketMaxOnlineCount * 3))
            .setThreadFactory(ThreadFactoryBuilder.create().setNamePrefix("Websocket-Pool-").build())
            .build();

    /**
     * 用户集
     */
    private static Map<String, Session> USERS = new ConcurrentHashMap<String, Session>();

    /**
     * 存储用户
     *
     * @param key     唯一键
     * @param session 用户信息
     */
    public static void put(String key, Session session) {
        USERS.put(key, session);
    }

    /**
     * 移除用户
     *
     * @param session 用户信息
     * @return 移除结果
     */
    public static boolean remove(Session session) {
        String key = null;
        boolean flag = USERS.containsValue(session);
        if (flag) {
            Set<Map.Entry<String, Session>> entries = USERS.entrySet();
            for (Map.Entry<String, Session> entry : entries) {
                Session value = entry.getValue();
                if (value.equals(session)) {
                    key = entry.getKey();
                    break;
                }
            }
        } else {
            return true;
        }
        return remove(key);
    }

    /**
     * 移出用户
     *
     * @param key 键
     */
    public static boolean remove(String key) {
        LOGGER.info("\n 正在移出用户 - {}", key);
        Session remove = USERS.remove(key);
        if (remove != null) {
            boolean containsValue = USERS.containsValue(remove);
            LOGGER.info("\n 移出结果 - {}", containsValue ? "失败" : "成功");
            return containsValue;
        } else {
            return true;
        }
    }

    /**
     * 获取在线用户列表
     *
     * @return 返回用户集合
     */
    public static Map<String, Session> getUsers() {
        return USERS;
    }

    /**
     * 群发消息文本消息
     *
     * @param message 消息内容
     */
    public static void sendMessageToUsersByText(String message) {
        Collection<Session> values = USERS.values();
        for (Session value : values) {
            pool.submit(new Runnable() {
                @Override
                public void run() {
                    sendMessageToUserByText(value, message);
                }
            });
        }
    }

    /**
     * 发送文本消息
     *
     * @param userName 自己的用户名
     * @param message  消息内容
     */
    public static void sendMessageToUserByText(Session session, String message) {
        if (session != null) {
            try {
                synchronized (session) {
                    session.getBasicRemote().sendText(message);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        } else {
            LOGGER.info("\n[你已离线]");
        }
    }
}

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

        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.8.3</version>
        </dependency>

这块具体的说明可以参考

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

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

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

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

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

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

新建IMUsers存储所有的im用户

package com.chrisf.imextend;

import cn.hutool.core.thread.ExecutorBuilder;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.chrisf.sdk.protocal.Protocal;
import com.chrisf.sdk.utils.LocalSendHelper;
import io.netty.channel.Channel;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * im 客户端用户集
 *
 * @author badao
 */
public class ImUsers {

    private static ExecutorService pool = ExecutorBuilder.create()
            .setCorePoolSize(20)//初始20个线程
            .setMaxPoolSize(40)//最大40个线程
            .setWorkQueue(new LinkedBlockingQueue<>(60))//有界等待队列，最大等待数是60
            .setThreadFactory(ThreadFactoryBuilder.create().setNamePrefix("IM-Pool-").build())//设置线程前缀
            .build();

    /**
     * 用户集
     */
    private static Map<String, Channel> USERS = new ConcurrentHashMap<String, Channel>();

    /**
     * 存储用户
     *
     * @param key     唯一键
     * @param session 用户信息
     */
    public static void put(String key, Channel session) {
        USERS.put(key, session);
    }

    /**
     * 移除用户
     *
     * @param session 用户信息
     * @return 移除结果
     */
    public static boolean remove(Channel session) {
        String key = null;
        boolean flag = USERS.containsValue(session);
        if (flag) {
            Set<Map.Entry<String, Channel>> entries = USERS.entrySet();
            for (Map.Entry<String, Channel> entry : entries) {
                Channel value = entry.getValue();
                if (value.equals(session)) {
                    key = entry.getKey();
                    break;
                }
            }
        } else {
            return true;
        }
        return remove(key);
    }

    /**
     * 移出用户
     *
     * @param key 键
     */
    public static boolean remove(String key) {
        Channel remove = USERS.remove(key);
        if (remove != null) {
            boolean containsValue = USERS.containsValue(remove);
            return containsValue;
        } else {
            return true;
        }
    }

    /**
     * 获取在线用户列表
     *
     * @return 返回用户集合
     */
    public static Map<String, Channel> getUsers() {
        return USERS;
    }

    /**
     * 群发消息文本消息
     *
     * @param protocal 消息内容
     */
    public static void sendMessageToUsersByText(Protocal protocal) {
        Collection<Channel> values = USERS.values();
        for (Channel value : values) {
            pool.submit(() -> sendMessageToUserByText(value, protocal));
        }
    }

    /**
     * 发送消息
     *
     * @param session
     * @param protocal
     */
    public static void sendMessageToUserByText(Channel session, Protocal protocal) {
        if (session != null) {
            synchronized (session) {
                try {
                    LocalSendHelper.sendData(session,protocal,null);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        } else {

        }
    }
}

在登录回调中新增用户

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

在登出回调中移除用户

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

在需要群发时群发消息

  if(ImTypeUEnum.CHAT_ROOM.getCode() == typeu)
  {
   ImUsers.sendMessageToUsersByText(p);
  }

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/135903.html

赞 (0)
飞熊的头像飞熊bm
0 0

相关推荐

发表回复

登录后才能评论
极客之音——专业性很强的中文编程技术网站，欢迎收藏到浏览器，订阅我们！