SpringCloud專案從訊息中間件中獲取到訊息（SpringCloud整合Netty和RabbitMQ以及WebSocket）-有解無憂

一、外部環境搭建

發送訊息到MQ和外部環境的搭建見上一章Springcloud專案發送訊息大RabbitMQ以及環境搭建
（注：RabbitMQ是安裝在虛擬機上的）

二、依賴注入

本文不僅匯入了上文的amqp依賴坐標還有新的netty依賴坐標

三、撰寫組態檔（yaml）

和上文一樣，不變的是這個，注意埠是5672,路徑看rabbitMQ安裝在本機還是虛擬機
在這里插入圖片描述

四、業務層邏輯分析

首先宣告本文的業務邏輯，各位讀者可能遇到的業務邏輯不一樣，所以寫法會有些許不同，但是大致還是一樣，本文在這先宣告本文在處理訊息發送時候的業務邏輯
業務場景：在用戶已經關注了粉絲的情況下，RabbitMQ中已經有了用戶的訊息佇列，那么我只需要在作者發布文章的時候或者點贊的時候，將存入進佇列的訊息立刻發送給已經登錄的用戶即可，
(注：發送訊息參考上文：發送訊息至MQ)

那么業務層的處理首先需要準備一下六個類：
在這里插入圖片描述
那么接下來就詳解每個類的作用，其中業務邏輯復雜的只有監聽器類和業務邏輯類
-------------------------------------------開始劃重點--------------------------------------------

## 以下是重點（重重重）

工具類“ApplicationContextProvider”：回傳一些需要的Bean實體以及背景關系物件實體（無需改變）

package com.tensquare.notice.config;

import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

@Component
public class ApplicationContextProvider implements ApplicationContextAware {
    /**
     * 背景關系物件實體
     */
    private static ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    /**
     * 獲取applicationContext
     *
     * @return
     */
    public static ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    /**
     * 通過name獲取 Bean.
     *
     * @param name
     * @return
     */
    public Object getBean(String name) {
        return getApplicationContext().getBean(name);
    }

    /**
     * 通過class獲取Bean.
     *
     * @param clazz
     * @param <T>
     * @return
     */
    public <T> T getBean(Class<T> clazz) {
        return getApplicationContext().getBean(clazz);
    }

    /**
     * 通過name,以及Clazz回傳指定的Bean
     *
     * @param name
     * @param clazz
     * @param <T>
     * @return
     */
    public <T> T getBean(String name, Class<T> clazz) {
        return getApplicationContext().getBean(name, clazz);
    }
}

Nettt服務類“NettyServer”：實作NIO的傳輸模式 --固定寫法，配置埠以及協議名即可（埠自定義，無需改變）

package com.tensquare.notice.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;

public class NettyServer {

    /**
     *  啟動netty服務，傳遞一個埠號
     */
    public void start(int port){
        System.out.println("準備啟動Netty......");
        //服務器引導程式
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        //用來處理新的連接
        EventLoopGroup boos = new NioEventLoopGroup();
        //用來處理業務邏輯（讀寫）
        EventLoopGroup worker = new NioEventLoopGroup();
        serverBootstrap.group(boos,worker)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                       //請求訊息解碼器
                        ch.pipeline().addLast(new HttpServerCodec());
                        //將多個訊息轉為單一的request或者response物件
                        ch.pipeline().addLast(new HttpObjectAggregator(65536));
                        //處理websocket的訊息事件(websocket服務器協議處理程式)
                        ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws"));
                        //創建自己的webscoket處理器，自己用來撰寫業務邏輯
                        MyWebSocketHandler myWebSocketHandler = new MyWebSocketHandler();
                        ch.pipeline().addLast(myWebSocketHandler);
                    }
                }).bind(port);
    }
}

Netty配置類“NettyConfig”：NettyConfig是Springcloud專案中的一種組態檔，自動加載，所以會自動開啟執行緒

因此需要configuration注解以及Bean注解

package com.tensquare.notice.config;

import com.tensquare.notice.netty.NettyServer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class NettyConfig {

    @Bean
    public NettyServer createNettyServer(){
        NettyServer nettyServer = new NettyServer();
        //啟動netty服務，使用新的執行緒啟動
        new Thread(){
            @Override
            public void run(){
                nettyServer.start(1234);
            }

        }.start();
        return nettyServer;
    }

}

訊息容器配置類：“RabbitConfig”類：宣告出需要的訊息容器，（注：與后續的訊息監聽器相呼應，名稱不建議改變）

package com.tensquare.notice.config;

import com.tensquare.notice.listener.SysNoticeListener;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

//配置類
@Configuration
public class RabbitConfig {

    @Bean("sysNoticeContainer")
    public SimpleMessageListenerContainer create(ConnectionFactory connectionFactory){
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        //使用Channel
        container.setExposeListenerChannel(true);
        //設定自己撰寫的監聽器
        container.setMessageListener(new SysNoticeListener());
        return container;
    }
}

通訊處理類“MyWebSocketHandler”類:也就是MQ和WebSocket進行互動

重點較多…
一：MyWebSocketHandler是用來進行通訊處理的，也就是MQ和WebSocket進行互動（通訊處理類–核心業務類）
二：MyWebSocketHandler進行業務處理，獲取訊息數量（業務場景：獲取到訊息數量即可）
三：MyWebSocketHandler繼承SimpleChannelInboundHandler< TextWebSocketFrame>，重寫channelRead0(ChannelHandlerContext 這個引數獲取連接,TextWebSocketFrame 這個引數獲取頁面引數

那么廢話不多說，先上代碼后上解釋：-----------

package com.tensquare.notice.netty;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.tensquare.entity.Result;
import com.tensquare.entity.StatusCode;
import com.tensquare.notice.config.ApplicationContextProvider;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.HashMap;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

//核心業務類，獲取MQ的訊息
public class MyWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    /**
     * 創建物件監聽器
     */
    private static ObjectMapper Mapper = new ObjectMapper();
    /**
     * 從Spring容器中獲取訊息監聽器容器，處理訂閱訊息sysNotice
     */
    SimpleMessageListenerContainer sysNoticeContainer = (SimpleMessageListenerContainer) ApplicationContextProvider.getApplicationContext().getBean("sysNoticeContainer");
    /**
     * 從spring容器中獲取RabbitTemplate
     *
     */
    RabbitTemplate rabbitTemplate = ApplicationContextProvider.getApplicationContext().getBean(RabbitTemplate.class);
   // @Autowired
   // private RabbitTemplate rabbitTemplate;
    /**
     * 存放WebScoket連接Map，根據用戶ID存放
     */
    public static ConcurrentHashMap<String, Channel> userChannelMap = new ConcurrentHashMap<>();
    /**
     *用戶請求服務端，執行的方法
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        //約定用戶第一次請求攜帶的資料:{"userid":"1"}
        //獲取用戶請求資料并決議
        String json = msg.text();
        //決議資料獲取用戶ID
        String userId = Mapper.readTree(json).get("userId").asText();
        //第一次請求的時候需要建立WebScoket連接
        Channel channel = userChannelMap.get(userId);
        if (channel==null){
            //獲取WebScoket連接
            channel  = ctx.channel();
            //把連接放到容器中
            userChannelMap.put(userId,channel);
        }
        //只用完成新訊息的提醒即可，只需要獲取訊息的數量
        //獲取RabbitMQ的內容，并且發送給用戶
        RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);
        //拼接捕獲佇列的名稱
        String queueName = "article_subscribe_"+userId;
        //獲取Rabbit的properties容器 （獲取rabbit的屬性容器）
        Properties queueProperties = rabbitAdmin.getQueueProperties(queueName);
        //獲取訊息數量
        int noticeCount = 0;
        //判斷properties是否不為空
        if (queueProperties!=null){
            //如果不為空，獲取訊息數量
            noticeCount = (int)queueProperties.get("QUEUE_MESSAGE_COUNT");
        }
        //----------------------------------
        //封裝回傳的資料
        HashMap countMap = new HashMap();
        countMap.put("sysNoticeCount",noticeCount);
        Result result = new Result(true, StatusCode.OK,"查詢成功！！",countMap);
        //把資料發送給用戶
        channel.writeAndFlush(new TextWebSocketFrame(Mapper.writeValueAsString(result)));
        //把訊息從佇列里清空，否則MQ訊息監聽器會再次消費一次
        if (noticeCount>0){
            rabbitAdmin.purgeQueue(queueName,true);
        }
        //為用戶的訊息佇列通知注冊監聽器，便于用戶在線的時候，
        //一旦有新訊息，可以主動推送給用戶，不需要用戶請求服務器獲取資料
        sysNoticeContainer.addQueueNames(queueName);
    }
}

**
接下來就是關于這個類的具體解釋了，務必細看，截圖都是從剛剛代碼中截取的，和我發的原始碼是一樣的

測驗引數是自定義的，真實開發環境不會如此
這個其實就是將引數獲取到，然后以id為標識將連接存入連接容器的程序
其中有一個Result類可以不用定義，本文是作測驗用的所以定義了
在這里插入圖片描述
通過管家獲取到訊息的數量

發送訊息的代碼

那么以上就是關于整個MyWebSocketHandler類的詳解，

監聽器：SysNoticeListener類：判斷用戶是否在線，發送訊息

package com.tensquare.notice.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.tensquare.entity.Result;
import com.tensquare.entity.StatusCode;
import com.tensquare.notice.netty.MyWebSocketHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;

import java.util.HashMap;

//訊息監聽器
public class SysNoticeListener implements ChannelAwareMessageListener {

    private static ObjectMapper MAPPER = new ObjectMapper();

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        //獲取用戶id，可以通過佇列名稱獲取
        String queueName = message.getMessageProperties().getConsumerQueue();
        String userId = queueName.substring(queueName.lastIndexOf("_")+1);
        io.netty.channel.Channel wsChannel = MyWebSocketHandler.userChannelMap.get(userId);
        //判斷用戶是否在線
        if (wsChannel!=null){
            //如果連接不為空，代表用戶在線
            //封裝回傳資料
            HashMap countMap = new HashMap();
            countMap.put("sysNoticeCount",1);
            Result result = new Result(true, StatusCode.OK,"查詢成功",countMap);
            //把資料通過WebScoket連接主動推送給用戶
            wsChannel.writeAndFlush(new TextWebSocketFrame(MAPPER.writeValueAsString(result)));
        }

    }

}

這里與RabbitConfig工具類中相對應
具體作用如注釋所說，
在這里插入圖片描述

測驗：這里將一個靜態html頁面用作測驗，加載服務的靜態資源里面即可

<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1transitional.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
<head>
    <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
    <title>測驗 notice 微服務與頁面 websocket 互動</title>
</head>
<body>
<h1>
    websocket連接服務器獲取mq訊息測驗
</h1>
<form onSubmit="return false;">
    <table><tr>
        <td><span>服務器地址：</span></td>
        <td><input type="text" id="serverUrl" value="ws://127.0.0.1:1234/ws" /></td>
      </tr>
      <tr>
        <td><input type="button" id="action" value="連接服務器" onClick="connect()" /></td>
        <td><input type="text" id="connStatus" value="未連接 ......" /></td>
    </tr></table>
    <br />
    <hr color="blue" />
    <div>
        <div style="width: 50%;float:left;">
            <div>
                <table><tr>
                    <td><h3>發送給服務端的訊息</h3></td>
                    <td><input type="button" value="發送" onClick="send(this.form.message.value)" /></td>
                </tr></table>
            </div>
            <div><textarea type="text" name="message" style="width:500px;height:300px;">
{
    "userId":"1"
}
                </textarea></div>
        </div>
        <div style="width: 50%;float:left;">
            <div><table>
                <tr>
                    <td><h3>服務端回傳的應答訊息</h3></td>
                </tr>
            </table></div>
            <div><textarea id="responseText" name="responseText" style="width: 500px;height: 300px;" onfocus="this.scrollTop = this.scrollHeight ">
這里顯示服務器推送的資訊
            </textarea></div>
        </div>
    </div>

</form>

<script type="text/javascript">
    var socket;
    var connStatus = document.getElementById('connStatus');;
    var respText = document.getElementById('responseText');
    var actionBtn = document.getElementById('action');
    var sysCount = 0;
    var userCount = 0;

    function connect() {
        connStatus.value = "正在連接 ......";

        if(!window.WebSocket){
            window.WebSocket = window.MozWebSocket;
        }
        if(window.WebSocket){

            socket = new WebSocket("ws://127.0.0.1:1234/ws");

            socket.onmessage = function(event){
                respText.scrollTop = respText.scrollHeight;
                respText.value += "\r\n" + event.data;
                var sysData = JSON.parse(event.data).data.sysNoticeCount;
                if(sysData){
                    sysCount = sysCount + parseInt(sysData)
                }
                var userData = JSON.parse(event.data).data.userNoticeCount;
                if(userData){
                    userCount = userCount + parseInt(sysData)
                }
                respText.value += "\r\n現在有" + sysCount + "條訂閱新訊息";
                respText.value += "\r\n現在有" + userCount + "條點贊新訊息";
                respText.scrollTop = respText.scrollHeight;
            };
            socket.onopen = function(event){
                respText.value += "\r\nWebSocket 已連接";
                respText.scrollTop = respText.scrollHeight;

                connStatus.value = "已連接 O(∩_∩)O";

                actionBtn.value = "斷開服務器";
                actionBtn.onclick =function(){
                    disconnect();
                };

            };
            socket.onclose = function(event){
                respText.value += "\r\n" + "WebSocket 已關閉";
                respText.scrollTop = respText.scrollHeight;

                connStatus.value = "已斷開";

                actionBtn.value = "連接服務器";
                actionBtn.onclick = function() {
                    connect();
                };
            };

        } else {
            //alert("您的瀏覽器不支持WebSocket協議！");
            connStatus.value = "您的瀏覽器不支持WebSocket協議！";
        }
    }

    function disconnect() {
        socket.close();
    }

    function send(message){
        if(!window.WebSocket){return;}
        if(socket.readyState == WebSocket.OPEN){
            socket.send(message);
        }else{
            alert("WebSocket 連接沒有建立成功！");
        }
    }
</script>
</body>
</html>

埠不需要改變，
下圖為測驗結果
在這里插入圖片描述

在這里插入圖片描述
可以看到，我多發送2條文章，由于關聯了一個粉絲，所以又多了2條訊息
而訊息中間件中訊息總數始終為01，因為都以及發送出去了

在這里插入圖片描述
結語：此篇與上一篇相呼應，有不對的地方望大佬指出，

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/242380.html

標籤：java

上一篇：SpringCloud OpenFeign 熔斷

下一篇：線性表