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

}

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