快捷導(dǎo)航

Java實(shí)現(xiàn)使用Websocket發(fā)送消息詳細(xì)代碼舉例

更新時間：2024年06月23日 11:19:19 作者：睡個好覺

這篇文章主要給大家介紹了關(guān)于Java實(shí)現(xiàn)使用Websocket發(fā)送消息的相關(guān)資料,WebSocket是一種協(xié)議,用于在Web應(yīng)用程序和服務(wù)器之間建立實(shí)時、雙向的通信連接,文中通過代碼介紹的非常詳細(xì),需要的朋友可以參考下

前言

需求背景：之前做了個樓棟房間數(shù)量出租了多少，需要在數(shù)據(jù)大屏上實(shí)時展示
解決方案：以切面攔截出租接口，當(dāng)有房間出租時重新查詢相關(guān)數(shù)據(jù)，封裝好用websocket實(shí)時傳送前端展示

一、websocket 簡介

WebSocket 是一種在單個TCP連接上進(jìn)行全雙工通信的協(xié)議。WebSocket 使得客戶端和服務(wù)器之間的數(shù)據(jù)交換變得更加簡單，允許服務(wù)端主動向客戶端推送數(shù)據(jù)。在WebSocket API中，瀏覽器和服務(wù)器只需要完成一次握手，兩者之間就直接可以創(chuàng)建持久性的連接，并進(jìn)行雙向數(shù)據(jù)傳輸。

二、websocket 如何使用

1.引入依賴包

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

2.創(chuàng)建 WebSocketConfig

package com.alpari.config;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

/**
 * @author wangrui
 * @date 2023/11/8
 **/
@Configuration
@EnableWebSocket
public class WebSocketConfig {

    @Bean
    public ServerEndpointExporter serverEndpoint() {
        return new ServerEndpointExporter();
    }

    /**
     * 注意事項(xiàng)：
     * WebSocket啟動的時候優(yōu)先于spring容器，從而導(dǎo)致在WebSocketServer中調(diào)用業(yè)務(wù)Service會報(bào)空指針異常
     * 所以需要在WebSocketServer中將所需要用到的service給靜態(tài)初始化一下
     * 此處提前注入
     */
//    @Autowired
//    private void setXxx(XxxService xxxService) {
//        WebSocketServer.xxxService = xxxService;
//    }
}

3.創(chuàng)建 WebSocketServer

package com.alpari.config;

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.CopyOnWriteArraySet;

/**
 * @author wangrui
 * @date 2023/11/8
 * @description: 若在此類中直接調(diào)用service方法失敗，需要加上：public static XxxService xxxService; 另外在config中注入實(shí)例
 **/
@Slf4j
@Component
@ServerEndpoint(value = "/websocket/{userId}")
public class WebSocketServer {

    /**
     * 當(dāng)前連接數(shù)
     */
    private static int onlineCount = 0;

    /**
     * concurrent包的線程安全Set，用來存放每個客戶端對應(yīng)的MyWebSocket對象
     */
    private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<>();

    /**
     * 與某個客戶端的連接會話，需要通過它來給客戶端發(fā)送數(shù)據(jù)
     */
    private Session session;

    /**
     * 接收的用戶id
     */
    private String userId;

    /**
     * 連接建立成功調(diào)用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        this.session = session;
        webSocketSet.add(this);
        addOnlineCount();        //在線數(shù)加1
        log.info("有新連接加入！當(dāng)前在線人數(shù)為" + getOnlineCount());
        this.userId = userId;
        try {
            sendMessage("連接成功");
        } catch (IOException e) {
            log.error("websocket IO異常", e);
        }
    }

    /**
     * 連接關(guān)閉調(diào)用的方法
     */
    @OnClose
    public void onClose() {
        webSocketSet.remove(this);
        subOnlineCount();               //在線數(shù)減1
        log.info("有一連接關(guān)閉！當(dāng)前在線人數(shù)為" + getOnlineCount());
    }

    /**
     * 收到客戶端消息后調(diào)用的方法
     *
     * @param message 客戶端發(fā)送過來的消息
     */
    @OnMessage
    public void onMessage(String message) {
        log.info("來自客戶端的消息:" + message);
        //群發(fā)消息
        for (WebSocketServer item : webSocketSet) {
            try {
                item.sendMessage(message);
            } catch (IOException e) {
                log.error("發(fā)送消息失敗", e);
            }
        }
    }

    /**
     * @param error 錯誤
     */
    @OnError
    public void onError( Throwable error) {
        log.error("發(fā)生錯誤", error);
    }

    /**
     * 發(fā)送消息
     * @param message 消息
     * @throws IOException 異常
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 群發(fā)自定義消息
     */
    public static void sendInfo(String message, @PathParam("userId") String userId) {
        log.info("消息內(nèi)容：" + message);
        for (WebSocketServer item : webSocketSet) {
            try {
                //這里可以設(shè)定只推送給這個sid的，為null則全部推送
                if (userId == null) {
                    item.sendMessage(message);
                } else if (item.userId.equals(userId)) {
                    item.sendMessage(message);
                }
            } catch (IOException e) {
                log.error("消息發(fā)送失敗", e);
            }
        }
    }

    /**
     * 獲取在線人數(shù)
     * @return 在線人數(shù)
     */
    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    /**
     * 在線人數(shù)加一
     */
    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    /**
     * 在線人數(shù)減一
     */
    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }
}

4.在線測試

在線ws測試地址：http://www.jsons.cn/websocket/

輸入測試

后端返回的結(jié)果

5.模擬消息發(fā)送

/**
 * 切面攔截?cái)?shù)據(jù)模擬給指定用戶或全部用戶發(fā)消息
 */
public void sendMessage(String userid) {
    // 此處模擬數(shù)據(jù)大屏推送，以切面方式攔截相關(guān)接口，若數(shù)據(jù)發(fā)生變化，則將數(shù)據(jù)推送，不傳id則全部發(fā)送
    String msg = "模擬數(shù)據(jù)";
    WebSocketServer.sendInfo(msg, userid);
    log.info("發(fā)送消息成功，消息內(nèi)容：{}，接收人：{}", msg, userid);
}