websocket实践

写在前面

此功能是基于分离版的若依系统实践，技术为springboot+websocket+vue，功能主要实现定时（每一分钟）给前台推送一条消息，前台显示推送的消息。

1. 后端

1.1 配置类（一般放在config文件夹下）

• ServerEndpointExporter 是一个类，属于 javax.websocket.server 包的一部分。
• 它的主要作用是自动扫描 Spring 容器中带有 @ServerEndpoint 注解的类，并将它们注册为 WebSocket 端点。
• 当我们在一个类上使用 @ServerEndpoint 注解时，必须通过某种方式将这个类注册到 WebSocket 容器中。ServerEndpointExporter 的作用就是自动完成这个注册过程。

package com.ruoyi.framework.config;

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

@Configuration
public class WebSocketConfig {

    /**
     * ServerEndpointExporter 是一个类，属于 javax.websocket.server 包的一部分。
     * 它的主要作用是自动扫描 Spring 容器中带有 @ServerEndpoint 注解的类，并将它们注册为 WebSocket 端点。
     * 当我们在一个类上使用 @ServerEndpoint 注解时，必须通过某种方式将这个类注册到 WebSocket 容器中。ServerEndpointExporter 的作用就是自动完成这个注册过程。
     * @return
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

1.2 构造的消息实体

package com.ruoyi.framework.ws.domain;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class MessageData {
    private String nickname;  // 发送者昵称
    private String content;   // 消息内容

}

1.3 工具类（辅助消息处理类）

SemaphoreUtils类：

• 信号量相关处理
• 在WebSocket 应用中，限制同时连接的客户端数量

package com.ruoyi.framework.ws.util;

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

/**
 * 信号量相关处理
 * 在WebSocket 应用中，限制同时连接的客户端数量
 * 
 */
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);
        }
    }
}

WebSocketUsers类

• websocket 客户端用户集
• 封装用户会话的管理、群发和单发消息操作，简化了 WebSocket 应用的开发

package com.ruoyi.framework.ws;

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 客户端用户集
 * 封装用户会话的管理、群发和单发消息操作，简化了 WebSocket 应用的开发
 *
 */
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 session 缓存
     * @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[你已离线]");
        }
    }
}

1.4 消息处理类

package com.ruoyi.framework.ws;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
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 com.ruoyi.framework.ws.util.SemaphoreUtils;
import com.ruoyi.framework.ws.util.WebSocketMessageScheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * websocket 消息处理
 *
 */
@Component
@ServerEndpoint("/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);

    private static Map<String, WebSocketMessageScheduler> schedulers = new ConcurrentHashMap<>();

    /**
     * 连接建立成功调用的方法
     */
    @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);
            WebSocketMessageScheduler scheduler = new WebSocketMessageScheduler(session);
            scheduler.start();
            schedulers.put(session.getId(), scheduler);
            LOGGER.info("\n 建立连接 - {}", session);
            LOGGER.info("\n 当前人数 - {}", WebSocketUsers.getUsers().size());
            WebSocketUsers.sendMessageToUserByText(session, "连接成功11111111");
        }
    }

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

    /**
     * 抛出异常时处理
     */
    @OnError
    public void onError(Session session, Throwable exception) throws Exception
    {
        if (session.isOpen())
        {
            // 关闭连接
            session.close();
        }
        String sessionId = session.getId();
        WebSocketMessageScheduler scheduler = schedulers.remove(sessionId);
        if (scheduler != null) {
            scheduler.stop();
        }
        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);
    }

}

2. 前端页面

在若依的菜单管理随便创建一个页面进行测试

<template>
  <div class="websocket-container">
    <h1>WebSocket 消息接收页面</h1>
    <div class="message-area">
      <h2>消息列表</h2>
      <ul class="message-list">
        <li v-for="(message, index) in messageList" :key="index" class="message-item">
          {{ message }}
        </li>
      </ul>
    </div>
    <div class="connection-status">
      <p>连接状态: {{ connectionStatus }}</p>
    </div>
    <button @click="closeWebSocket" class="close-button">关闭连接</button>
  </div>
</template>

<script>
export default {
  data() {
    return {
      ws: null,
      messageList: [],
      connectionStatus: '未连接'
    };
  },
  mounted() {
    this.initWebSocket();
  },
  beforeDestroy() {
    if (this.ws) {
      this.ws.close();
    }
  },
  methods: {
    initWebSocket() {
      const wsUrl = 'ws://127.0.0.1:8080/websocket/message'; // 替换为你的 WebSocket 服务器地址
      this.ws = new WebSocket(wsUrl);

      this.ws.onopen = () => {
        this.connectionStatus = '已连接';
        console.log('WebSocket 连接已建立');
      };

      this.ws.onerror = (error) => {
        this.connectionStatus = '连接错误';
        console.error('WebSocket 错误：', error);
      };

      this.ws.onmessage = (event) => {
        const message = event.data;
        this.messageList.push(message);
        console.log('接收到消息：', message);
      };

      this.ws.onclose = () => {
        this.connectionStatus = '已断开';
        console.log('WebSocket 连接已关闭');
      };
    },
    closeWebSocket() {
      if (this.ws && this.ws.readyState === WebSocket.OPEN) {
        this.ws.close();
        this.connectionStatus = '已手动关闭';
      }
    }
  }
};
</script>

<style scoped>
.websocket-container {
  max-width: 800px;
  margin: 0 auto;
  padding: 20px;
  font-family: Arial, sans-serif;
}

.message-area {
  margin-top: 20px;
}

.message-list {
  list-style-type: none;
  padding: 0;
  margin: 0;
  max-height: 400px;
  overflow-y: auto;
  border: 1px solid #ddd;
  border-radius: 4px;
}

.message-item {
  padding: 10px;
  margin-bottom: 5px;
  background-color: #f9f9f9;
  border-radius: 4px;
}

.connection-status {
  margin-top: 20px;
  padding: 10px;
  background-color: #e9f7ef;
  border-radius: 4px;
}

.close-button {
  margin-top: 20px;
  padding: 10px 20px;
  background-color: #4CAF50;
  color: white;
  border: none;
  border-radius: 4px;
  cursor: pointer;
}

.close-button:hover {
  background-color: #45a049;
}
</style>

3. 最终效果

4. 可能坑点

1. 如果和我一样基于若依，记得给/websocket/**放行（在SecurityConfig中）
   

2. 可以使用ApiFox测试websocket是否连接成功
   ApiFox有专门的测试websocket接口，记得加上token

5. 拓展

基于这个的成功，我紧接着实现了一个基于websocket简单的聊天应用，效果图如下：

总结

写这篇一是自己加深记忆，其次就是共享。

写在最后

如果此文对您有所帮助，请帅戈靓女们务必不要吝啬你们的Zan，感谢！！不懂的可以在评论区评论，有空会及时回复。