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;
    }

}

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