WebSocket双工通信实现用户互踢功能，一个用户同时只能在一台设备上登录需求服务端实现

用户3587585

发布于 2024-06-13 19:03:12

510

发布于 2024-06-13 19:03:12

引言

最近有个需求需要控制用户在登录系统时一个用户只能在一台设备上登录。如果用户已经在一台设备上登录了，然后同一个用户又继续使用另一台设备登录，则需要踢掉在前一台设备上登录的会话，确保一个用户同一时间只有一个会话。笔者在掘金上调研了可行的技术方案，发现主要有以下两种实现方案：

一、客户端向服务端轮询获取当前登录用户信息，具体步骤如下：

1）用户登录成功后在浏览器的localStorage中保存用户的userId和sessionId（即会话ID，当用户每次在后台登录成功后生成一个uuid代表sessionId）, 同时服务端也同时保存这些信息，如果用户在别的设备上登录则根据userId更新sessionId；
2）客户端通过一个定时器根据userId向服务端轮询获取当前用户最新的登录信息, 如果发现获取到的sessionId与本地localStorage中保存的sessionId不一致时就说明用户已经在别的设备上登录，则需要使当前会话失效，并跳转到用户登录页面

二、通过WebSocket技术实现，具体步骤如下：

1）用户在服务端登录成功后生成一个uuid代表sessionId，并返回给客户端；
2）客户端拿到服务端返回的sessionId后向服务端建立一个WebSocket连接，并使用一个HashMap数据结果存储sessionId与WebSocket的映射关系，同时使用Redis分布式数据库存储userId与sessionId列表的映射关系；
3）用户在一台设备上登录成功后，首先根据userId这个key去redis中查询当前userId对应的sessionId列表中是否已经存在一个sessionId。如果存在则根据这个sessionId从存储sessionId与WebSocket映射关系的HashMap中找到对应的WebSocket会话实例，并发送消息给客户端通知当前用户已在别的设备上登录，当前会话失效；
4）客户端收到WebSocket推送过来的服务端会话已失效通知后清除浏览器本地缓存localStorage和会话缓存sessionStorage中保存的变量，然后跳转到用户登录页面。

对于第一种方案客户端向服务端轮询获取当前登录用户的sessionId方式，懂行的人一眼就看得出来比较耗费服务器的资源和网络带宽，而且定时间间隔时间设置长了还无法实时感知到当前用户已经在别的设备上登录，况且用户也不会经常有这种同时在两台设备上登录的行为。显然这种方案不是一个很好的解决方案。

而对于第二种方案通过WebSocket双工通信的方式就优越的多，它不需要客户端向服务端轮询获取用户的sessionId，而且当用户同时在两台设备上登录时主动推送消息给前一个登录的客户端通知当前会话已失效即可。那么我选择WebScoket技术实现这一需求也就水到渠成了。因为笔者之前也从未体验过WebSocket双工通信，那么本文就带大家使用WebSocket+Redis技术实现这一具有挑战意义的需求。

由于同时实现这一需求的前后端功能篇幅太长，笔者把它分为两篇文章写完，本文我们着眼于服务端功能的实现，下一篇文章笔者再实现客户端功能，并对我们要实现的功能进行效果测试。

WebSocket简介

这里，我们通过向最近很火的chatGPT提问 WebSocket，根据chatGP回答内容简单介绍以下什么是WebSocket？

翻译过来就是：

“WebSocket是一种向客户端和服务端提供双向、低延时和实时通信的通信>协议，它是出现是为了克服传统的Http连接基于请求响应模式、不能在服>务端和客户端提供持久的连接等一些缺陷。 ”
“WebSocket 具备全双工通信能力，意味着服务端和客户端都可以在任意时>间向对方发送消息，无需向另一方发起请求。这与传统的Http连接每次获>得服务端的响应信息都必须在客户端发起一次请求完全不同。 ”
“WebSocket使用单独的TCP连接用于通信，它可以确保连接在需要的时候一>直打开，这有效减少了为了每次请求/响应建立和保持的多个连接造成的>>服务负担。 ”
“WebSocket 目前广泛用于在线游戏、社交聊天和实时股票更新等应用场>>>景。大多数现代Web浏览器都已经支持WebSocket，并且能在HTML、JavaSc>rip和CSS等前端技术中一起使用。 ”

重点：WebSocket是一种可以在服务端和客户端实现双工通信的通信协议，它客服了Http通信协议的客户端每次向服务端获取数据必须依次经过建立http会话连接、发起请求、等待服务端响应等流程的弊端。WebSocket通信协议可以让客户端和服务端在需要的期间保持长连接，并在建立会话连接后任意时刻向对方发送消息。

新建项目并添加依赖

使用IDEA新建一个SpringBoot项目，并在项目的pom.xml文件中加入web-mvc、spring-security、mysql驱动、mybatis-plus和web-socket等项目业务需要的maven依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.7.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.bonus</groupId>
    <artifactId>bonus-backend</artifactId>
    <version>1.0.0-SNAPSHOT</version>
    <name>bonus-backend</name>
    <description>bonus-backend</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <!--spring web mvc依赖-->
       <dependency>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-starter-web</artifactId>
       </dependency>
        <!--spring security安全框架-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-security</artifactId>
            <version>2.2.7.RELEASE</version>
        </dependency>
        <!--spring配置注解自动生效依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>
        <!--阿里fastjson-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.79</version>
        </dependency>
        <!--mysql 驱动-->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.16</version>
            <scope>runtime</scope>
        </dependency>
        <!--druid数据源-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid-spring-boot-starter</artifactId>
            <version>1.2.8</version>
        </dependency>
        <!--mybatis-plus-->
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>3.1.2</version>
        </dependency>
        <!--WebSocket依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
            <version>2.2.7.RELEASE</version>
        </dependency>
        <!--redis客户端升级工具redisson-->
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.10.7</version>
        </dependency>
        <!--代码简洁工具lombok-->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>2.2.7.RELEASE</version>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

新建ServerEndpointExporter和RedissonClient配置Bean

WebSocketConfig.java

package com.bonus.bonusbackend.config;

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.codec.JsonJacksonCodec;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

@Configuration
public  class WebSocketConfig {
    @Value("${spring.redis.host}")
    private String redisHost;
    @Value("${spring.redis.port}")
    private String redisPort;

    /**
     * WebSocket 服务端点导出bean
     * @return
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter(){
        return new ServerEndpointExporter();
    }

    /**
     * Redisson 客户端，单机模式
     * @return RedissonClient instance
     */
    @Bean
    public RedissonClient redissonSingle(){
        Config config = new Config();
        config.setCodec(new JsonJacksonCodec())
                .useSingleServer()
                .setAddress("redis://127.0.0.1:6379");
        return Redisson.create(config);
    }

}

ServerEndpointExporter类bean用来暴露websocket服务端点，RedissonClient类bean则是用来操作redis的客户端工具。

新建WebSocketServer组件类

新建WebSocketServer组件类，并完成与客户端websocket的打开会话onOpen、收到消息onMessage、关闭会话onClose和会话出错onError等事件监听方法WebSocketServer.java

package com.bonus.bonusbackend.config;

import javax.annotation.Resource;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

@Component
@ServerEndpoint(value = "/wsMessage")
public  class WebSocketServer {
    /**
    * 存放当前连接的客户端WebSocket会话对象
    */
    private  static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
    /**与客户的连接会话，通过它来给客户端发送数据*/
    private Session session;
    /**当前会话id*/
    private String sessionId;
    /**在线用户集合*/
    private  static List<String> memAccounts = new ArrayList<>();
    /**日志打印器*/
    private  static  final Logger log = LoggerFactory.getLogger(WebSocketServer.class);
    
    /**
     * 连接打开时调用
     * @param session 会话
     */
    @OnOpen
    public void onOpen(Session session) throws InterruptedException, IOException {
        log.info("打开webSocket会话");
        this.session = session;
        log.info("queryString:{}", session.getQueryString());
        String queryStr = session.getQueryString();
        JSONObject queryJson = assembleQueryJson(queryStr);
        /**这里因项目业务需要memAccount代替userId*/
        String memAccount = queryJson.getString("memAccount");
        String sessionId = queryJson.getString("sessionId");
        this.memAccount = memAccount;
        this.sessionId = sessionId;
        // 判断会话是否已经存在
        if(webSocketMap.containsKey(sessionId)){
            // 已存在，先移除后添加
            webSocketMap.remove(sessionId);
            webSocketMap.put(sessionId, this);
        } else{
            // 不存在直接添加
            webSocketMap.put(sessionId, this);
            // 增加在线人数
            if(!memAccounts.contains(memAccount)){
                memAccounts.add(memAccount);
            }
            String message = genMessage(200, "当前在线人数为："+getOnlineNumber());
            sendMessageAll(message);
        }
        log.info("连接用户："+memAccount+",当前连接人数为："+getOnlineNumber());
    }
    
    /**
     * 生成消息
     * @param code
     * @param message
     * @return
     */
    public String genMessage(int code, String message){
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("code", code);
        jsonObject.put("msg", message);
        return jsonObject.toJSONString();
    }
    
    /**获取当前在线人数，暂时没有考虑分布式*/
    public synchronized int getOnlineNumber() {
        return this.memAccounts.size();
    }
    
    /**从查询参数中提取用户账号和会话ID参数*/
    public JSONObject assembleQueryJson(String queryStr){
        JSONObject jsonObject = new JSONObject();
        if(StringUtils.isEmpty(queryStr)){
            return jsonObject;
        }
        String[] queryParams = queryStr.split("&");
        for (int i=0;i< queryParams.length;i++){
                String[] nameValues = queryParams[i].split("=");
                if("memAccount".equals(nameValues[0])){
                    memAccount = nameValues[1];
                }else if("sessionId".equals(nameValues[0])){
                    sessionId = nameValues[1];
                }
        }
        jsonObject.put("memAccount", memAccount);
        jsonObject.put("sessionId", sessionId);
        return jsonObject;
    }
    
    @OnMessage
    public void onMessage(String message, Session session){
        log.info("收到客户端webSocket消息:{}, queryString={}", message, session.getQueryString());
    }
    
    @OnClose
    public void onClose(Session session) throws InterruptedException, IOException {
        log.info("webSocket关闭会话");
        String queryStr = session.getQueryString();
        log.info("queryStr:{}", queryStr);
        if(webSocketMap.containsKey(sessionId)){
            webSocketMap.remove(sessionId);
            memAccounts.remove(memAccount);
            log.info("用户:"+memAccount+"退出系统,当前连接人数为："+getOnlineNumber());
        }
        /**群发在线用户*/
        String message = genMessage(200, "当前连接人数为："+getOnlineNumber());
        sendMessageAll(message);
    }
    
    /**
     * 单发消息给客户端
     * @param message 消息内容
     * @param sessionId 会话id
     * @throws IOException
     */
    public void sendMessage(String message, String sessionId) throws IOException {
        log.info("发送消息给"+sessionId+",内容为："+message);
        if(StringUtils.isNotEmpty(sessionId) && webSocketMap.containsKey(sessionId)){
            webSocketMap.get(sessionId).sendMessage(message);
        }
    }
    
    /**
     * 单发消息
     * 服务端向客户端发送消息
     * @param message
     * @throws IOException
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 群发消息
     * @param message
     * @throws IOException
     */
    public void sendMessageAll(String message) throws IOException {
        for(String sessionId: webSocketMap.keySet()){
            webSocketMap.get(sessionId).sendMessage(message);
        }
    }
    
     @OnError
    public void onError(Session session, Throwable ex){
        log.error("webSocket Error", ex);
        JSONObject queryJson = assembleQueryJson(session.getQueryString());
        String sessionId = queryJson.getString("sessionId");
        log.error(sessionId+"连接出错", ex);
    }
      
}

新增踢出之前登录用户方法

在WebSocketServer类中新增判断同一用户是否有超过1个会话，如果有则踢出前面的会话

        // 同一个账号允许的最大会话数    private static int MAX_SESSION_SIZE = 1;    
    /**redis中存储用户的key前缀*/
    
    private  static  String USER_KEY_PREFIX = "memInfo_";
    
    /**用户websocket会话队列前缀*/
    private static String USER_DEQUE_PREFIX = "memInfo_deque_";
    /**判断同一用户是否存在多个会话并踢出前一个会话时的锁前缀*/
    private static String LOCK_KEY_PREFIX = "memInfo_lock_";

/**
     * 判断用户是否存在多台设备登录，若存在则踢掉前面登录的用户
     * @param sessionId 会话ID
     * @param memAccount 会员账户
     */
    public void kickOut(String sessionId, String memAccount){
        log.info("sessionId:{},memAccount:{}", sessionId, memAccount);
        String userKey = USER_KEY_PREFIX+memAccount+"_"+sessionId;
        RBucket<JSONObject> redisBucket = redissonClient.getBucket(userKey);
        JSONObject currentUser = redisBucket.get();
        log.info("currentUser={}", JSONObject.toJSONString(currentUser));
        String userDequeKey = USER_DEQUE_PREFIX+currentUser.getString("memAccount");
        // 锁定
        String lockKey = LOCK_KEY_PREFIX+memAccount;
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(2L, TimeUnit.SECONDS);
        try {
            RDeque<String> deque = redissonClient.getDeque(userDequeKey);
            // 若队列中没有此sessionId,且用户未被踢出则加入队列
            if(!deque.contains(sessionId) && !currentUser.getBoolean("isKickOut")){
                deque.push(sessionId);
            }
            // 若队列里的sessionId数超过最大会话数，则开始踢用户
            while (deque.size()>MAX_SESSION_SIZE){
                String kickOutSessionId;
                if(KICKOUT_AFTER){
                    kickOutSessionId = deque.removeFirst();
                } else {
                    kickOutSessionId = deque.removeLast();
                }
                RBucket<JSONObject> kickOutBucket = redissonClient.getBucket(USER_KEY_PREFIX+memAccount+"_"+kickOutSessionId);
                JSONObject kickOutUser = kickOutBucket.get();
                if(kickOutUser!=null){
                    kickOutUser.put("isKickOut", true);
                    log.info("kickOutUser={}", kickOutUser.toJSONString());
                    kickOutBucket.set(kickOutUser);
                    JSONObject wsJson = new JSONObject();
                    wsJson.put("code", 1001); // 响应码为1001代表被踢出登录
                    wsJson.put("msg", "本账号别处登录或被踢出，如有疑问请联系上级");
                    sendMessage(wsJson.toJSONString(), kickOutSessionId);
                    currentUser = redisBucket.get();
                }
            }
            if(currentUser.getBoolean("isKickOut")){
                JSONObject wsJson = new JSONObject();
                wsJson.put("code", 1001);
                wsJson.put("msg", "本账号别处登录或被踢出，如有疑问请联系上级");
                sendMessage(wsJson.toJSONString(), this.sessionId);
            }
        } catch (Exception e){
            log.error("kickOut error", e);
        } finally {
            // 释放锁
            if(lock.isHeldByCurrentThread()){
                lock.unlock();
                log.info("用户："+memAccount+" unlock");
            }else{
                log.info("用户："+memAccount+" already release lock ");
            }
        }

    }

服务端用户登录成功后逻辑

用户登录成功后调用WebSocketServer#kickout方法异步判断当前登录用户是否存在多个会话，若存在则踢掉前一个会话

这异步逻辑在Security配置类的configure(HttpSecurity http)方法的登录成功处理器中完成

如何在spring-security框架中实现用户登录逻辑网上已经有太多文章，这里就不赘述了，读者也可以参考笔者之前发布的文章Spring Security入门（三）: 基于自定义数据库查询的认证实战

SecurityConfig.java

    @Resource
    private RedissonClient redissonClient;

    @Resource(name = "asyncThreadPool")
    private ThreadPoolExecutor poolExecutor;

    @Resource
    private WebSocketServer webSocketServer;
protected void configure(HttpSecurity http) throws Exception {
    // 不拦截websocket通信，因为websocket是在认证成功之后进行的
    http.authorizeRequests().antMatchers("/wsMessage").permitAll()
        .anyRequest().authenticated()
                .and().httpBasic()
                .and().formLogin()
                .loginProcessingUrl("/member/login")
                 .successHandler((httpServletRequest, httpServletResponse, authentication) -> {
                     httpServletResponse.setContentType("application/json;charset=utf-8");
                     httpServletResponse.setStatus(HttpStatus.OK.value());
                     PrintWriter printWriter = httpServletResponse.getWriter();
                     // 从authentication入参中获取认证信息memInfoDTO（代表用户信息）
                     MemInfoDTO memInfoDTO = (MemInfoDTO) authentication.getPrincipal();
                     Map<String, Object>
                         userJson.put("memAccount", memInfoDTO.getMemAccount());
                     userJson.put("sessionId", sessionId);
                     userJson.put("isKickOut", false);
                     RBucket<JSONObject> bucket = redissonClient.getBucket("memInfo_"+memInfoDTO.getMemAccount()+"_"+sessionId);
                     bucket.set(userJson, 2*60*60L, TimeUnit.SECONDS);userMap = new HashMap<>();
                     userMap.put("memAccount", memInfoDTO.getMemAccount());
                     // 生成一个uuid作为会话id
                     String sessionId = UUID.randomUUID(true).toString().replaceAll("-","");               dataMap.put("sessionId", sessionId);
                     // 将用户信息保存到redis中：key为memInfo_+ memAccount + _ + sessionId
                     JSONObject userJson = new JSONObject();
                     userJson.put("memAccount", memInfoDTO.getMemAccount());
                     userJson.put("sessionId", sessionId);
                     userJson.put("isKickOut", false); // 是否被踢掉标识
                     // 将用户账号和会话等信息保存在redis中，过期时间2小时
                     RBucket<JSONObject> bucket = redissonClient.getBucket("memInfo_"+memInfoDTO.getMemAccount()+"_"+sessionId);
                     bucket.set(userJson, 2*60*60L, TimeUnit.SECONDS);
                      // 异步判断用户是否在其他设备上已登录，若是则踢掉前一个登录的用户
                     poolExecutor.execute(()->{
                        webSocketServer.kickOut(sessionId, memInfoDTO.getMemAccount());
                     });
                     // 给客户端返回用户信息和会话id
                      ResponseResult<Map<String, Object>> responseResult = ResponseResult.success(dataMap, "login success");
                     printWriter.write(JSONObject.toJSONString(responseResult));
                     printWriter.flush();
                     printWriter.close();
                 });
        
    
}

异步线程池配置

package com.bonus.bonusbackend.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Configuration
public  class TheadPoolConfig {

    @Bean("asyncThreadPool")
    public ThreadPoolExecutor threadPoolExecutor(){
        BlockingDeque<Runnable> blockingDeque = new LinkedBlockingDeque<>(100);
        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5, 10, 60, TimeUnit.SECONDS, blockingDeque);
        return poolExecutor;
    }
}

启动类加上@EnableAsync注解

@SpringBootApplication
@EnableAsync
public class BonusBackendApplication {
    public static void main(String[] args) {
        SpringApplication.run(BonusBackendApplication.class, args);
    }
}

项目环境变量

各个环境共享环境配置变量文件:

application.properties

server.servlet.context-path=/bonus
spring.profiles.active=dev
spring.jackson.date-format=yyyy-MM-dd HH:mm:ss
spring.jackson.time-zone=GMT+8

开发环境配置变量文件:

application-dev.properties

server.address=127.0.0.1
server.port=8090

spring.datasource.druid.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.druid.url=jdbc:mysql://localhost:3306/bonus?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&allowMultiQueries=true
spring.datasource.druid.username=bonus_user
spring.datasource.druid.password=tiger2022@
spring.datasource.druid.validation-query=select 1 from dual

然后运行启动类的main方法就能给客户端提供服务了。