1. AI技术聚合首页
  2. Python

集成websocket实现实时通信(ruoyi 使用笔记)

社会演员多 • • Python • 阅读 85

集成websocket实现实时通信(ruoyi 使用笔记

  • 1.简单介绍WebSocket
  • 2.详细代码
    • 2.1WebSocketConfig
    • 2.2 SemaphoreUtils
    • 2.3 WebSocketServer
    • 2.4 WebSocketUsers工具类
    • 2.5 html
    • 2.6 vue版本前端代码
    • 2.7 controller

1.简单介绍WebSocket

Websocket 是一种基于 TCP 协议的全双工通信协议，它使得客户端和服务器之间可以进行实时的双向通信。相对于传统的 HTTP 协议只能通过客户端发送请求，然后等待服务端的响应，WebSocket 可以让客户端和服务器在任何时候都可以相互发送消息，这种实时通信的方式非常适合需要实时更新数据的应用场景，比如聊天室、在线游戏、股票行情等。

WebSocket 的运作流程如下：

客户端向服务器发起 WebSocket 握手请求；
服务器返回确认信息给客户端，完成握手；
握手成功后，客户端和服务器就可以通信了；
双方可以随时发送消息到对方，也可以关闭连接。

WebSocket 的好处包括：

实时性：WebSocket 提供了双向通信能力，可以实现实时更新数据的功能；
可靠性：WebSocket 基于 TCP 协议，可以保证消息传输的可靠性；
性能高：WebSocket 的开销小，通信效率高，不会频繁地进行连接、断开等操作，降低网络延迟；
跨域支持：WebSocket 支持跨域通信，可以在不同的域之间建立连接。

在前端开发中，使用 WebSocket 可以使用 JavaScript WebSocket API 来进行操作，常见的库包括 Socket.io 和 WebSocket-Node 等。在后端开发中，WebSocket 的实现可以使用 Node.js、Java、Python 等多种语言和框架。

当然，在使用 WebSocket 时也需要考虑一些安全问题，比如避免跨站脚本攻击（XSS）、防范恶意请求等。

若依网址点击即可；

链接: https://pan.baidu.com/s/13JVC9jm-Dp9PfHdDDylLCQ 提取码: y9jt

2.详细代码

（大家可以去若依下载我自己做下记录而已）

2.1WebSocketConfig

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();
    }
}

2.2 SemaphoreUtils

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.3 WebSocketServer



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.lxh.demo.util.SemaphoreUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * websocket 消息处理
 * 
 * @author ruoyi
 */
@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);

    /**
     * 连接建立成功调用的方法
     */
    @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);
    }
}

2.4 WebSocketUsers工具类

提供一些简单的工具


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 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[你已离线]");
        }
    }
}

2.5 html

<!DOCTYPE html>
<html lang="zh" xmlns:th="http://www.thymeleaf.org">
<head>
    <meta charset="utf-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <title>测试界面</title>
</head>

<body>

<div>
    <input type="text" style="width: 20%" value="ws://127.0.0.1/websocket/message" id="url">
	<button id="btn_join">连接</button>
	<button id="btn_exit">断开</button>
</div>
<br/>
<textarea id="message" cols="100" rows="9"></textarea> <button id="btn_send">发送消息</button>
<br/>
<br/>
<textarea id="text_content" readonly="readonly" cols="100" rows="9"></textarea>返回内容
<br/>
<br/>
<script th:src="@{/js/jquery.min.js}" ></script>
<script type="text/javascript">
    $(document).ready(function(){
        var ws = null;
        // 连接
        $('#btn_join').click(function() {
        	var url = $("#url").val();
            ws = new WebSocket(url);
            ws.onopen = function(event) {
                $('#text_content').append('已经打开连接!' + '\n');
            }
            ws.onmessage = function(event) {
                $('#text_content').append(event.data + '\n');
            }
            ws.onclose = function(event) {
                $('#text_content').append('已经关闭连接!' + '\n');
            }
        });
        // 发送消息
        $('#btn_send').click(function() {
            var message = $('#message').val();
            if (ws) {
                ws.send(message);
            } else {
                alert("未连接到服务器");
            }
        });
        //断开
        $('#btn_exit').click(function() {
            if (ws) {
                ws.close();
                ws = null;
            }
        });
    })
</script>
</body>
</html>

2.6 vue版本前端代码

url: “ws://localhost:8080/websocket
我们本地配置为：ws://127.0.0.1:18080/websocket/message
ws/wss 不可修改，ip为本地需要连接地址ip websocket/message 为默认调用ruoyi中WebSocketServer 接口；

大家有使用网关之类的需要配置网关的地址；前段去访问网关的全部的地址即可；

微服务项目记得写端口我这边必须写不然访问不到 可能因为么走网关的原因；

//data中定义变量
lockReconnect: false,
      wsCfg: {
        // websocket地址(需要更改连接地址)
        url: "ws://localhost:8080/websocket
},
//method中定义方法

 createWebSocket() {
      console.log('createWebSocket')
            try {
                // 创建Web Socket 连接
                socket = new WebSocket(this.wsCfg.url);
                // 初始化事件
                this.initEventHandle(socket);
            } catch (e) {
                // 出错时重新连接
                console.log(e)
                this.reconnect(this.wsCfg.url);
            }
        },
    initEventHandle(socket) {
            // 连接关闭时触发
            socket.onclose = () => {
                console.log("连接关闭");
            };
            // 通信发生错误时触发
            socket.onerror = () => {
                // 重新创建长连接
                this.reconnect();
            };
            // 连接建立时触发
            socket.onopen = () => {
                console.log("连接成功");
            };
            // 客户端接收服务端数据时触发
            socket.onmessage = msg => {
                // 业务逻辑处理
              try{
                this.websocketSetFlag =	JSON.parse(msg.data).content;
                console.log(this.websocketSetFlag )
              }catch{

              }
						  
            };
        },
    reconnect() {
            //重连
            if (this.lockReconnect) {
                return;
            }
            this.lockReconnect = true;
            // 没连接上会一直重连，设置延迟避免请求过多
            setTimeout(() => {
                this.lockReconnect = false;
                this.createWebSocket(this.wsCfg.url);
            }, 2000);
        },
    sendWebsocket(text){
        //发送数据
        socket.send(
        JSON.stringify({
            type: "control",
            content: text,
        })
      ) ;
    },

2.7 controller

@Controller
@RequestMapping("/user")
public class HelloController {

    // websocket测试
    @GetMapping("/ws")
    public String ws()
    {
        return "websocket";
    }
}


嵌入微服务项目我是直接把他直接放在固定模块中的；


此文章暂时只适用于新手；

扩展：单个多个连接这个文章也不错

文章出处登录后可见！

已经登录？立即刷新

共计人评分，平均分

到目前为止还没有投票！成为第一位评论此文章。

赞 (0)
社会演员多的头像社会演员多普通用户
0
上一篇 2023年6月10日
下一篇 2023年6月10日

相关推荐