websocket整合rabbitMQ——實現消息推送

一、實現功能：當前頁面點擊請求后，另一個頁面立即收到一條消息，用springboot+websocket+rabbitMQ實現。

二、java代碼：

1、先在pom里導入需要的坐標

        <!--rabbitMQ-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
            <version>2.1.4.RELEASE</version>
        </dependency>

        <!--webSocket-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
            <version>2.1.4.RELEASE</version>
        </dependency>

<!--lombok，可用於log輸出到控制台（和System.out.println()一樣）-->
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <version>2.1.4.RELEASE</version>
</dependency>

 

2、編寫websocket配置類（這個配置除了監聽隊列的名字外，以后回來直接Ctrl+C）

@Slf4j
@ServerEndpoint("/websocket")//把當前類標識成一個WebSocket的服務端，值是訪問的URL地址
@Component//spring注入
public class WebSocketServer {

    //與某個客戶端的連接會話，需要通過它來給客戶端發送數據
    private Session session;

    //存放session的集合，很重要！！
    private static CopyOnWriteArraySet<WebSocketServer> sessions = new CopyOnWriteArraySet<WebSocketServer>();

    //用於存所有的連接服務的客戶端，這個對象存儲是安全的（因為HashMap不支持線程同步）
    private static ConcurrentHashMap<String, WebSocketServer> webSocketSet = new ConcurrentHashMap<>();


    //監聽隊列，從隊列中把剛發送的消息取出來
    @RabbitListener(queues = "pointQueue")
    public void getMessAge(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {

        System.out.println("發來消息：" + message);

        channel.basicAck(tag, false);//確認收到--消費

        //用來判斷session中是否存在數據
        if (sessions.size() != 0) {
            for (WebSocketServer s : sessions) {
                if (s != null) {
                    s.session.getBasicRemote().sendText(message);//向已連接客戶端發送信息
                }
            }
        }
    }


    //連接成功
    @OnOpen
    public void onOpen(Session session) {
        this.session = session;
        //這個一定要寫，第一次很容易忽略！
        sessions.add(this);
        log.info("[WebSocket] 連接成功，當前連接人數為：={}", webSocketSet.size());

    }

    //連接斷開
    @OnClose
    public void onClose() {
        //釋放
        sessions.remove(this);
        log.info("[WebSocket] 退出成功，當前連接人數為：={}", webSocketSet.size());
    }

    //收到消息
    @OnMessage
    public String onMessage(String message) {
        log.info("[WebSocket] 收到消息：{}", message);

        //這里是自己測試的，可以忽略
        if (message.equals("一只羊")) {//不能用==
            return "1:1個大白羊";
        }
        if (message.equals("兩只羊")) {
            return "2:2個大白羊";
        }
        if (message.equals("三只羊")) {
            return "3:個大白羊";
        }

        return "你已成功連接，這是webSocket服務端的返回信息！";
    }

}

 

3、rabbitMQ創建隊列與交換機綁定，這里不記錄了，主要是在里面加上這一條

    //websocket 放入到spring容器
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

 

二、Html代碼：注意這里是接收頁面的js代碼！！！

<script type="text/javascript">

    //設置一個變量，用於存儲websocket
    var socket = null;

    //判斷當前瀏覽器是否支持WebSocket
    if ('WebSocket' in window) {
        //alert('當前瀏覽器支持WebSocket')
        //實現化WebSocket對象，指定要連接的服務器地址與端口  建立連接
        //這個url必須寫對！ws固定的，ip是自己配置的，路徑是自己命名的
        socket = new WebSocket("ws://127.0.0.1:7777/websocket");
    }
    else {
        alert('當前瀏覽器 Not support websocket')
    }


    //連接成功（回調方法）
    socket.onopen = function () {
        alert("Socket 連接成功");
    }

    //連接錯誤（回調方法）
    socket.onerror = function () {
        alert('連接出錯')
        //可嘗試刷新頁面--循環
        //window.location.reload();
    };

    //接收消息（回調方法）
    socket.onmessage = function (event) {
        alert("你收到了一條返回消息："+event.data);
    }

    //連接關閉（回調方法）
    socket.onclose = function () {
        alert("連接關閉");
    }

    //窗口關閉，關閉websocket（否則server端會拋異常）
    window.onbeforeunload = function () {
        socket.close();
    }
    

</script>

 

 

四、總結：

到這里最核心的東西已經搭建完了，但肯定沒有寫全，不是自己不想寫，因為一些簡單枯燥的東西不必要羅列出來，否則雜多不知其意。

所以放在最后用最簡潔的語言，描述大量的簡單代碼是最高效的：

1. 首先創建一個發送消息的html頁面，用簡單的http請求向一個controller，如ajax/axios，把想要發送的消息作為參數發送過去。
2. controller接收到消息參數后，調用rabbitMQ的方法放入隊列，此時候監聽這個隊列的方法立馬取出隊列中的消息（並消費）。
3. 監聽隊列方法中，用session.getBasicRemote().sendText("你的消息"); 發送給已連接websocket的接收頁面，這時頁面就會alert()彈出你發送頁面發送的消息。