SpringBoot基于WebSocket进行推送

Posted 2023-04-23

参考技术A

客户端发起http请求，请求Netty服务器进行WebSocket连接，服务器接收后请求后进行注册信道并登记客户端IP地址，如此一来就建立了WebSocket通讯连接。

上面的论述可以得出，我们可以比较Http和WebSocket两者之间的关系和区别

基于上述WebSocket的特点，可以实现在线设备的消息推送

创建线程启动Netty服务器

也许这里，有些疑问为什么需要创建线程来启动Netty服务器，这里尝试一下如果没有创建线程，直接在Main主线程启动，看看会有什么问题？这个问题留着后面再去说明

在配置文件中进行WebSocket的端口配置

上面的步骤进行完成了一个Netty服务器的初始化启动和端口监听，这时候会有疑问了，消息推送功能还没有实现。所以下面就是我们如何去实现消息处理逻辑

springboot 项目==基于websocket的服务端推送消息。

1.创建springboot项目，首先我们还是先引入依赖

<!-- webSocket  begin-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<!-- webSocket  end-->

2.创建配置类   WebSocketH5Config

@Configuration
@EnableWebSocket
public class WebSocketH5Config implements WebSocketConfigurer {
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
   //handler是webSocket的核心，配置入口
   registry.addHandler(new MyHandler(), "/myHandler/{ID}").setAllowedOrigins("*").addInterceptors(new WebSocketInterceptor());
    }

3.创建服务类   MyHandler

@Service
public class MyHandler implements WebSocketHandler {

    //在线用户列表
    private static final Map<String, WebSocketSession> users;

    static {
        users = new HashMap<>();
    }

    //新增socket
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        String ID = session.getUri().toString().split("ID=")[1];
        //使用URLDecoder进行解码，因为前段传来的就是乱码
        String decode = URLDecoder.decode(ID, "UTF-8");
        System.out.println("成功建立连接,注册账号为" + decode);
        if (decode != null) {
            users.put(decode, session);
            session.sendMessage(new TextMessage("成功建立socket连接"));
            System.out.println(session);
        }
        System.out.println("当前在线人数：" + users.size());
    }

    //接收socket信息(接受客户端的信息)
    @Override
    public void handleMessage(WebSocketSession webSocketSession, WebSocketMessage<?> webSocketMessage) throws Exception {
        try {
            JSONObject jsonobject = JSONObject.fromObject(webSocketMessage.getPayload());
//            System.out.println(jsonobject.get("id"));
//            System.out.println(jsonobject.get("message") + ":来自" + (String) webSocketSession.getAttributes().get("WEBSOCKET_USERID") + "的消息");
            Map map = new HashMap(2);
            map.put("id", jsonobject.get("id"));
            map.put("message", jsonobject.get("message").toString());
            String msg = JSON.toJSONString(map);
            //将消息发送到mq
            Sender been = (Sender) AppContext.getBeen(Sender.class);
            been.send(msg);
            sendMessageToUser(jsonobject.get("id") + "", new TextMessage(msg));
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 发送信息给指定用户 （服务端给指定用户发送消息）
     *
     * @param clientId
     * @param message
     * @return
     */
    public boolean sendMessageToUser(String clientId, TextMessage message) {
        if (users.get(clientId) == null) return false;
        WebSocketSession session = users.get(clientId);
//        System.out.println("sendMessage:" + session);
        if (!session.isOpen()) return false;
        try {
            session.sendMessage(message);
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    /**
     * 广播信息（服务端给所有用户推送消息）
     *
     * @param message
     * @return
     */
    public boolean sendMessageToAllUsers(TextMessage message) {
        boolean allSendSuccess = true;
        Set<String> clientIds = users.keySet();
        WebSocketSession session = null;
        for (String clientId : clientIds) {
            try {
                session = users.get(clientId);
                if (session.isOpen()) {
                    session.sendMessage(message);
                }
            } catch (IOException e) {
                e.printStackTrace();
                allSendSuccess = false;
            }
        }

        return allSendSuccess;
    }


    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        if (session.isOpen()) {
            session.close();
        }
        System.out.println("连接出错");
        users.remove(getClientId(session));
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        System.out.println("连接已关闭：" + status);
        users.remove(getClientId(session));
    }

    @Override
    public boolean supportsPartialMessages() {
        return false;
    }

    /**
     * 获取用户标识
     *
     * @param session
     * @return
     */
    private Integer getClientId(WebSocketSession session) {
        try {
            Integer clientId = (Integer) session.getAttributes().get("WEBSOCKET_USERID");
            return clientId;
        } catch (Exception e) {
            return null;
        }
    }

    public Map<String, WebSocketSession> getAllUsers() {
        return users;
    }
}

4.创建配置类   WebSocketInterceptor

 

public class WebSocketInterceptor implements HandshakeInterceptor {

    //在握手之前执行该方法, 继续握手返回true, 中断握手返回false. 通过attributes参数设置WebSocketSession的属性
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse serverHttpResponse, WebSocketHandler webSocketHandler, Map<String, Object> attributes) throws Exception {
        if (request instanceof ServletServerHttpRequest) {
            String ID = request.getURI().toString().split("ID=")[1];
            //使用URLDecoder进行解码，因为前段传来的就是乱码
            String decode = URLDecoder.decode(ID, "UTF-8");
            System.out.println("当前session的ID=" + decode);
            //ServletServerHttpRequest serverHttpRequest = (ServletServerHttpRequest) request;
            //HttpSession session = serverHttpRequest.getServletRequest().getSession();
            attributes.put("WEBSOCKET_USERID", decode);
        }
        return true;
    }

    //完成握手之后执行该方法
    @Override
    public void afterHandshake(ServerHttpRequest serverHttpRequest, ServerHttpResponse serverHttpResponse, WebSocketHandler webSocketHandler, Exception e) {
        System.out.println("进来webSocket的afterHandshake拦截器！");
    }
}

5.jsp页面

<%@ page language="java" contentType="text/html; charset=UTF-8" pageEncoding="UTF-8" %>
<!DOCTYPE html>
<html>
<head>
    <title>socket.html</title>

    <meta name="keywords" content="keyword1,keyword2,keyword3">
    <meta name="description" content="this is my page">
    <%--<meta name="content-type" content="text/html" charset="UTF-8">--%>
    <%--<meta http-equiv="Content-Type" content="text/html; charset=UTF-8">--%>
    <!--<link rel="stylesheet" type="text/css" href="./styles.css">-->

</head>

<body>

Welcome<br/>

<input id=‘ipt‘ type=‘text‘ name="注册"/>
<button onclick="getVal()">注册</button>
<br>
<input id="text" type="text"/>
<button onclick="">发送的内容</button>
<br>
<input id=‘ipts‘ type=‘text‘ name="发给谁"/>
<button onclick="send()">发给谁</button>
<button onclick="closeWebSocket()">断开连接</button>

</div>
<!-- 公共JS -->
 
<script src="https://code.jquery.com/jquery-3.3.1.min.js"></script>
<script type="text/javascript" src="../webSocket/jquery.min.js"></script>


<script type="text/javascript">

    var websocket = null;

    function getVal() {
        //创建WebSocket
        connectWebSocket();
    }

    //强制关闭浏览器  调用websocket.close（）,进行正常关闭
    window.onunload = function () {

        //关闭连接
        closeWebSocket();
    }

    //建立WebSocket连接
    function connectWebSocket() {
        var user = document.getElementById(‘ipt‘);//找到id是ipt的input节点;
        var userID = user.value;
        // alert(userID);
        console.log("开始...");

        //建立webSocket连接
        websocket = new WebSocket("ws://192.168.1.46:9999/myHandler/ID=" + userID);
        // websocket = new WebSocket("ws://94.191.34.57:9999/myHandler/ID=" + userID);

        //打开webSokcet连接时，回调该函数
        websocket.onopen = function () {
            console.log("onpen");
        }

        //关闭webSocket连接时，回调该函数
        websocket.onclose = function () {
            //关闭连接
            console.log("onclose");
        }

        //接收信息
        websocket.onmessage = function (msg) {
            console.log("receive msg");
            console.log(msg.data);
            alert(msg.data)
        }
    }

    //发送消息
    function send() {
        var user = document.getElementById(‘ipts‘);//找到id是ipt的input节点;
        var userID = user.value;
        // alert(userID);
        var postValue = {};
        postValue.id = userID;
        postValue.message = $("#text").val();
        websocket.send(JSON.stringify(postValue));
    }

    //关闭连接
    function closeWebSocket() {
        if (websocket != null) {
            websocket.close();
        }
    }
</script>
<script type="text/javascript" charset="UTF-8"></script>
</body>
</html>