springboot?使用websocket技術(shù)主動給前端發(fā)送消息的實現(xiàn)

更新時間：2021年12月20日 08:46:35 作者：yueF_L

這篇文章主要介紹了springboot?使用websocket技術(shù)主動給前端發(fā)送消息的實現(xiàn)方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教

使用websocket技術(shù)主動給前端發(fā)送消息

springBoot2.0對WebSocket的支持簡直太棒了，直接就有包可以引入

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

WebSocketConfig

啟用WebSocket的支持也是很簡單

package com.spark.common.config; 
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
 
/**
 * @Author Lxq
 * @Date 2021-06-12 17:11
 * @Version 1.0
 * 開啟websocket支持
 */
@Configuration
public class WebSocketConfig {
 
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

WebSocketServer

這就是重點了，核心都在這里。

因為WebSocket是類似客戶端服務端的形式(采用ws協(xié)議)，那么這里的WebSocketServer其實就相當于一個ws協(xié)議的Controller

直接@ServerEndpoint("/socketServer/{userId}") 、@Component啟用即可，然后在里面實現(xiàn)@OnOpen開啟連接，@onClose關(guān)閉連接，@onMessage接收消息等方法。

新建一個ConcurrentHashMap webSocketMap 用于接收當前userId的WebSocket，方便IM之間對userId進行推送消息。單機版實現(xiàn)到這里就可以。

package com.spark.common.utils.websocket; 
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.spark.common.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component; 
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
 
/**
 * @Author Lxq
 * @Date 2021-06-12 17:13
 * @Version 1.0
 */
@ServerEndpoint("/socketServer/{userId}")
@Component
@Slf4j
public class WebSocketServer { 
    /**
     * 靜態(tài)變量，用來記錄當前在線連接數(shù)。應該把它設(shè)計成線程安全的。
     */
    private static int onlineCount = 0;
    /**
     * concurrent包的線程安全Set，用來存放每個客戶端對應的MyWebSocket對象。
     */
    private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
    /**
     * 與某個客戶端的連接會話，需要通過它來給客戶端發(fā)送數(shù)據(jù)
     */
    private Session session;
    /**
     * 接收userId
     */
    private String userId = "";
 
    /**
     * 連接建立成功調(diào)用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        this.session = session;
        this.userId = userId;
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            webSocketMap.put(userId, this);
            //加入set中
        } else {
            webSocketMap.put(userId, this);
            //加入set中
            addOnlineCount();
            //在線數(shù)加1
        }
 
        log.info("用戶連接:" + userId + ",當前在線人數(shù)為:" + getOnlineCount());
 
        try {
            sendMessage("連接成功");
        } catch (IOException e) {
            log.error("用戶:" + userId + ",網(wǎng)絡異常!");
        }
    } 
 
    /**
     * 連接關(guān)閉調(diào)用的方法
     */
    @OnClose
    public void onClose() {
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            //從set中刪除
            subOnlineCount();
        }
        log.info("用戶退出:" + userId + ",當前在線人數(shù)為:" + getOnlineCount());
    } 
 
    /**
     * 收到客戶端消息后調(diào)用的方法
     *
     * @param message 客戶端發(fā)送過來的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("用戶消息:" + userId + ",報文:" + message);
        //可以群發(fā)消息
        //消息保存到數(shù)據(jù)庫、redis
        if (StringUtils.isNotBlank(message)) {
            try {
                //解析發(fā)送的報文
                JSONObject jsonObject = JSON.parseObject(message);
                //追加發(fā)送人(防止串改)
                jsonObject.put("fromUserId", this.userId);
                String toUserId = jsonObject.getString("toUserId");
                //傳送給對應toUserId用戶的websocket
                if (StringUtils.isNotBlank(toUserId) && webSocketMap.containsKey(toUserId)) {
                    webSocketMap.get(toUserId).sendMessage(jsonObject.toJSONString());
                } else {
                    log.error("請求的userId:" + toUserId + "不在該服務器上");
                    //否則不在這個服務器上，發(fā)送到mysql或者redis
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
 
    /**
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("用戶錯誤:" + this.userId + ",原因:" + error.getMessage());
        error.printStackTrace();
    }
 
    /**
     * 實現(xiàn)服務器主動推送
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    } 
 
    /**
     * 發(fā)送自定義消息
     */
    public static void sendInfo(String message, @PathParam("userId") String userId) throws IOException {
        log.info("發(fā)送消息到:" + userId + "，報文:" + message);
        if (StringUtils.isNotBlank(userId) && webSocketMap.containsKey(userId)) {
            webSocketMap.get(userId).sendMessage(message);
        } else {
            log.error("用戶" + userId + ",不在線！");
        }
    }
 
    /**
     * 獲取當前在線人數(shù)
     *
     * @return
     */
    public static synchronized int getOnlineCount() {
        return onlineCount;
    }
 
    /**
     * 添加人數(shù)
     */
    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }
 
    /**
     * 減少人數(shù)
     */
    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    } 
}

websocket基礎(chǔ)入門-前端發(fā)送消息

項目結(jié)構(gòu)如下圖

TestSocket.java

package com.charles.socket; 
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint; 
@ServerEndpoint(value = "/helloSocket")
public class TestSocket { 
    /***
     * 當建立鏈接時，調(diào)用的方法.
     * @param session
     */
    @OnOpen
    public void open(Session session) {        
        System.out.println("開始建立了鏈接...");
        System.out.println("當前session的id是：" + session.getId());
    }
    
    /***
     * 處理消息的方法.
     * @param session
     */
    @OnMessage
    public void message(Session session, String data) {        
        System.out.println("開始處理消息...");
        System.out.println("當前session的id是：" + session.getId());
        System.out.println("從前端頁面?zhèn)鬟^來的數(shù)據(jù)是：" + data);
    }
}

index.jsp 代碼如下：

<%@ page language="java" contentType="text/html; charset=UTF-8" pageEncoding="UTF-8"%>
<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
<html>
<head>
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
<title>Charles-WebSocket</title> 
<script type="text/javascript">    
    var websocket = null;
    var target = "ws://localhost:8080/websocket/helloSocket";    
    function buildConnection() {
        
        if('WebSocket' in window) {
            websocket = new WebSocket(target);        
        } else if('MozWebSocket' in window) {
            websocket = MozWebSocket(target);
        } else {
            window.alert("瀏覽器不支持WebSocket");
        }
    }
    
    // 往后臺服務器發(fā)送消息.
    function sendMessage() {        
        var sendmsg = document.getElementById("sendMsg").value;
        console.log("發(fā)送的消息：" + sendmsg);        
        // 發(fā)送至后臺服務器中.
        websocket.send(sendmsg);
    }
    
</script>
</head>
<body>
    
    <button onclick="buildConnection();">開始建立鏈接</button>
    <hr>
    <input id="sendMsg" /> <button onclick="sendMessage();">消息發(fā)送</button> 
</body>
</html>

注意：

和后臺交互的時候，一定要先點擊：開始建立連接。你懂的...沒有建立連接的話，是不能發(fā)送消息的。