RabbitMQ最核心的交換機和佇列Exchange、Queue詳解-有解無憂

引言

最近公司專案中，車輛大資料的推送和接收同步都用到了RabbitMQ訊息中間件，對于其中最核心的交換機和佇列Exchange、Queue的引數配置和使用，再此簡單總結一下，供自己和大家一塊兒學習！

1.先來介紹RabbitMQ中的成員

Producer(生產者)：將訊息發送到Exchange
Exchange(交換器)：將從生產者接收到的訊息路由到Queue
Queue(佇列)：存放供消費者消費的訊息
BindingKey(系結鍵)：建立Exchange與Queue之間的關系（個人看作是一種規則，也就是Exchange將什么樣的訊息路由到Queue）
RoutingKey(路由鍵)：Producer發送訊息與路由鍵給Exchange，Exchange將判斷RoutingKey是否符合BindingKey，如何則將該訊息路由到系結的Queue
Consumer(消費者)：從Queue中獲取訊息

下面是各個成員的作用圖解

引入依賴

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.6.0</version>
</dependency>

2.先來介紹Exchange

這里將著重于介紹Exchange和Queue的各個引數解釋

先來看看Exchange中都有哪些屬性

exchange：名稱
type：型別
durable：是否持久化，RabbitMQ關閉后，沒有持久化的Exchange將被清除
autoDelete：是否自動洗掉，如果沒有與之系結的Queue，直接洗掉
internal：是否內置的，如果為true，只能通過Exchange到Exchange
arguments：結構化引數

下面這個類用于創建一個與RabbitMQ的Connection(連接)，該Connection用于創建Channel(信道)，Channel是訊息讀寫的通道，也就是我們的操作都會在Channel的基礎之上進行

2.1先使用最簡單的引數構建Exchange
exchangeDeclare(String exchange, String type)

進入RabbitMQ可視化界面可以看到，RabbitMQ已經為我們創建了exchange.0，型別為direct

具體釋意

  name                      名稱
  type                        型別
  Features                 特征
  Message rate in       訊息速率輸入
  Message rate out     訊息速率輸出

2.2接下來是三個引數，也就是加上了是否持久化，同時保留先前兩個引數的exchange.0，之前我們已經創建了exchange.0，那么我們再創建一次會怎樣

exchangeDeclare(String exchange, String type, boolean durable)

運行成功，并沒有報錯，因為只要你設定的的設定是一樣的，那么就不會報錯，如果設定的不一樣，那么就會報錯，后面會進行驗證

這里我們發現exchange.2多了一個D標識，這個D是durable也就是持久化，而exchange.0沒有持久化，也就是默認非持久化

接下來驗證這個持久化有什么作用
關閉rabbitmq
rabbitmqctl stop_app
啟動rabbitmq
rabbitmqctl start_app
重新進入可視化界面，Exchange就只剩下持久化的了

2.3接下來是五個引數的

多了兩個引數，autoDelete和arguments
exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, Map<String, Object> arguments)

下面創建了兩個Exchange
exchange.3自動洗掉為false
exchange.4自動洗掉為true
由于這里是沒有系結Queue的，那么exchange.4將在創建后就被洗掉掉？

執行上面的代碼

exchange.4還活的好好的，這是因為我們必須在系結Queue之后再失去系結才會被洗掉，否則為什么不直接拋例外，接下來進行驗證
下面直接通過可視化工具創建一個名稱為queue.4的Queue

英文釋義

Name         名稱
Features     特征
Status        狀態
Ready        是否準備好
Unacked     未確認
Total           總計
incoming     進來的
deliver        傳送
get             得到
ack             確認

2.5講解完Exchange的引數，再來看Queue的引數，就會發現只有一個exclusive未講
queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
Map<String, Object> arguments

exclusive：是否排他，如果未true，則只在第一次創建它的Connection中有效，當Connection關閉，該Queue也會被洗掉

在執行完下面代碼，查看可視化界面，發現queue中并沒有exclusive.queue，因為在connection關閉后，該queue也會自動洗掉

創建實體

package com.tiandy.illegal.util.mq;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.*;
import com.tiandy.illegal.bo.CLS_ManageService;
import com.tiandy.illegal.bo.CLS_ManageServiceImpl;
import com.tiandy.illegal.util.CLS_ILLEGAL_Error;
import com.tiandy.illegal.vo.CLS_VO_Message;
import com.tiandy.illegal.vo.CLS_VO_Record;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.ResourceBundle;

public class RabbitMQSend {
    //rabbitmq連接
    public static Connection connection = null;
    //rabbitmq通道
    public static Channel channel = null;
    //連接狀態標識
    public static boolean connectStatus = false;
    // 配置
    static ResourceBundle resourceBundle = ResourceBundle.getBundle("mq/artemisConfig");
    // 交換機  exchangeTemp
    private static String rabbitmq_exchange = resourceBundle.getString("rabbitmq_exchange");
    // 佇列名  queue_vbs_vehicle_record
    private static String rabbitmq_queue = resourceBundle.getString("rabbitmq_queue");
    // service
    CLS_ManageService cls_manageService = new CLS_ManageServiceImpl();
    static ConnectionFactory factory = null;

    public void initialize() {
        try {
            //連接工廠
            if (null == factory) {
                factory = new ConnectionFactory();
                factory= RabbitMQUtil.getRabbitMQConnectionFactory();
                // 關閉通道與連接
                closeConnection();
                connection = factory.newConnection();
                channel = connection.createChannel();
                // 宣告交換機
                // channel.exchangeDeclare(rabbitmq_exchange, BuiltinExchangeType.DIRECT ,true);
                connectStatus = true;

            }
        } catch (Exception e) {
            connectStatus = false;
            e.printStackTrace();
           // log.error("RabbitMQSend method initialize:" + e.getMessage(), e);
        }
    }

    //關閉連接
    public void closeConnection() {
        try {
            if (channel != null) {
                if (channel.isOpen()) {
                    channel.close();
                    channel = null;
                }
            }
        } catch (Exception e) {
            //log.error("RabbitMQSend closeChannel  error  " + e);
            e.printStackTrace();
        }
        try {
            if (connection != null) {
                if (connection.isOpen()) {
                    connection.close();
                    connection = null;
                }
            }
        } catch (Exception e) {
           // log.error("RabbitMQSend closeConnection  error  " + e);
            e.printStackTrace();
        }
    }
/**
     * 監聽訊息佇列，獲取資料
     */
    public void queueDeclareExchange() {
        //宣告交換機
        try {
            Map<String, Object> args = new HashMap<String, Object>();
            args.put("x-max-length", 100000); // 設定最大存盤訊息數
            // 宣告交換機  (交換機引數)
            channel.exchangeDeclare(rabbitmq_exchange, BuiltinExchangeType.FANOUT, true);
            // 訊息持久化  (佇列引數)
            channel.queueDeclare(rabbitmq_queue, true, false, false, args);
            // 交換機與佇列系結
            channel.queueBind(rabbitmq_queue, rabbitmq_exchange, "");
            // 消費者限制
            //channel.basicQos(1);
            Consumer consumer = new DefaultConsumer(channel) {
                int inRecord=0; // 插入記錄數量
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //接收到的訊息
                    String msg = new String(body, "UTF-8");
                    // 判斷資料是否允許接入
                    int check = checkMessage(msg);
                    if (check == CLS_ILLEGAL_Error.ERROR_OK) {
                        // 訊息轉換至VO
                        CLS_VO_Message msgVo = cls_manageService.getMessageVo(msg);
                        // 判斷資料，分開處理白車牌資料與其他資料,每次新增一條
                        int count = cls_manageService.decideData(msgVo);
                        if(count>0){
                            inRecord+=count;
                            System.out.println("  已消費訊息："+envelope.getDeliveryTag()+"  插入記錄數：" + inRecord);
                        }
                    }
                    // 單條訊息確認（第幾條，是否多條）
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            // 設定訊息手動確認 （佇列名，是否自動確認，consumer）
            channel.basicConsume(rabbitmq_queue, false, consumer);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    /**
     * 方法說明：監測接收資訊
     *
     * @param message
     * @return @修改人及日期： @修改描述： @其他：
     */
    public int checkMessage(String message) {
        // TODO 監測資料格式及是否允許接入
        int check = 0;
        CLS_VO_Message vo_Message = null;
        try {
            vo_Message = JSONObject.parseObject(message, CLS_VO_Message.class);
        } catch (Exception e) {
            return CLS_ILLEGAL_Error.ERROR_PARAM;
        }
        if (vo_Message.getStorage_id() == null || "".equals(vo_Message.getStorage_id())) {
            return CLS_ILLEGAL_Error.ERROR_PARAM;
        }
        if (vo_Message.getCap_pic() == null || vo_Message.getCap_pic().size() == 0) {
            return CLS_ILLEGAL_Error.ERROR_PARAM;
        }
        if (vo_Message.getTotal_info() == null) {
            return CLS_ILLEGAL_Error.ERROR_PARAM;
        }
        CLS_VO_Record total_info = vo_Message.getTotal_info();
        if (total_info.getTollgateID() == null || "".equals(total_info.getTollgateID())) {
            return CLS_ILLEGAL_Error.ERROR_PARAM;
        }
        return check;
    }

}

至此，簡單的引數講解和應用就總結完了！

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

標籤：Java

上一篇：8. 老板不加薪，我用了這篇加了 3K

下一篇：Java 記憶體模型 JMM 詳解！