RabbitMQ詳解（上）

①：生產者（Producer）
    產生資料傳送訊息的程式是生產者
②：交換機（Exchange）
    交換機是 RabbitMQ 非常重要的一個部件，一方面它接收來自生產者的訊息，另一方面它將訊息推播到佇列中。交換機必須確切知道如何
　　處理它接收到的訊息，是將這些訊息推播到特定佇列還是推播到多個佇列，亦或者是把訊息丟棄，這個得有交換機型別決定
③：佇列（Queue）
    佇列是 RabbitMQ 內部使用的一種資料結構，儘管訊息流經 RabbitMQ 和應用程式，但它們只能儲存在佇列中。佇列僅受主機的記憶體和
　　磁碟限制的約束，本質上是一個大的訊息緩衝區。許多生產者可以將訊息傳送到一個佇列，許多消費者可以嘗試從一個佇列接收資料。這就是
　　我們使用佇列的方式
④：消費者（Consumer）
    消費與接收具有相似的含義。消費者大多時候是一個等待接收訊息的程式。請注意生產者，消費者和訊息中介軟體很多時候並不在同一機器上。
　　同一個應用程式既可以是生產者又是可以是消費者。

Broker：
    接收和分發訊息的應用，RabbitMQ Server就是Message Broker；簡單來說就是訊息佇列伺服器實體Virtual host：出於多租戶和
　　 安全因素設計的，把AMQP的基本元件劃分到一個虛擬的分組中，類似於網路中的namespace概念。當多個不同的使用者使用同一個
　　 RabbitMQ server提供的服務時，可以劃分出多個vhost，每個使用者在自己的vhost建立exchange／queue等
Connection：
    publisher／consumer和broker之間的TCP連線
Channel：
    訊息通道，如果每一次存取RabbitMQ都建立一個Connection，在訊息量大的時候建立TCPConnection的開銷將是巨大的，效率也較
　　 低。Channel是在connection內部建立的邏輯連線，如果應用程式支援多執行緒，通常每個thread建立單獨的channel進行通訊，AMQP
　　 method包含了channel id幫助使用者端和message broker識別channel，所以channel之間是完全隔離的。Channel作為輕量級的
　　 Connection極大減少了作業系統建立TCP connection的開銷。
Exchange：
　　message到達broker的第一站，根據分發規則，匹配查詢表中的routing key，分發訊息到queue中去。
　　常用的型別有：direct (point-to-point), topic (publish-subscribe) and fanout(multicast)
Queue：
    訊息佇列載體，每個訊息都會被投入到一個或多個佇列;訊息最終被送到這裡等待consumer取走
Routing Key：
    路由關鍵字，exchange根據這個關鍵字進行訊息投遞。
Binding：
　　exchange和queue之間的虛擬連線，binding中可以包含routing key，Binding資訊被儲存到exchange中的查詢表中，用於message
　　的分發依據;它的作用就是把exchange和queue按照路由規則繫結起來。
producer：
    訊息生產者，就是投遞訊息的程式。
consumer：
    訊息消費者，就是接受訊息的程式。

# 安裝 erlang 環境
rpm -ivh erlang-23.3.4.11-1.el7.x86_64.rpm
# 安裝 socat 環境
yum -y install socat
# 安裝 RabbitMQ 服務
rpm -ivh rabbitmq-server-3.9.13-1.el7.noarch.rpm
# 檢查是否安裝
yum list | grep rabbitmq
yum list | grep erlang

補充命令：
# 檢視所以的已開啟的埠
firewall-cmd --zone=public --list-ports
# 開啟15672埠（--permanent代表永久生效，重啟系統不會失效）
firewall-cmd --zone=public --add-port=15672/tcp --permanent

# 防火牆關閉
systemctl stop firewalld
# 啟動RabbitMQ服務
systemctl start rabbitmq-server
    或 /sbin/service rabbitmq-server start
    或 service rabbitmq-server start
# 新增開機啟動RabbitMQ服務
chkconfig rabbitmq-server on 
# 開啟web管理介面（可以更方便快速的對RabbitMQ進行操作）
rabbitmq-plugins enable rabbitmq_management
# 停止RabbitMQ服務
systemctl stop rabbitmq-server
    或 /sbin/service rabbitmq-server stop
    或 service rabbitmq-server stop

注：web管理介面應用的操作
rabbitmqctl stop_app　　停止web頁面
rabbitmqctl start_app  啟動web頁面

實際操作說明：
rabbitmqctl list_users
    # 檢視RabbitMQ裡的所有使用者
rabbitmqctl list_vhosts
    # 檢視RabbitMQ裡的所有vhosts
rabbitmqctl list_permissions
    # 檢視RabbitMQ裡所有使用者的許可權
rabbitmqctl list_user_permissions guest
    # 檢視RabbitMQ裡guest使用者的許可權

rabbitmqctl add_vhost test
    # 建立的一個虛擬主機項為 test 的名稱
rabbitmqctl add_user admin 123
    # 建立一個使用者為admin 密碼為123
rabbitmqctl set_user_tags admin administrator
    # 設定admin的角色為超級管理員（administrator）
rabbitmqctl set_permissions -p test admin ".*" ".*" ".*"
    # 設定admin在test的vhost中，並設定全部檔案的讀寫操作 
rabbitmqctl list_permissions -p test
    # 檢視test中的vhost裡的使用者

systemctl stop rabbitmq-server
    # 停止RabbitMQ服務
yum list | grep rabbitmq
    # 檢視RabbitMQ安裝的相關列表
yum -y remove rabbitmq-server.noarch
    # 解除安裝RabbitMQ已安裝的相關內容
yum list | grep erlang
    # 檢視erlang安裝的相關列表
yum -y remove erlang-*
yum remove erlang.x86_64
    # 解除安裝erlang已安裝的相關內容
rm -rf /usr/lib64/erlang 
rm -rf /var/lib/rabbitmq
rm -rf /usr/local/erlang
rm -rf /usr/local/rabbitmq
    # 刪除有關的所有檔案

package cn.xw.helloWorld;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2022-03-04 17:42
 */
public class Producer {
    //簡單佇列名稱
    public static final String QUEUE_NAME = "helloWorldQueue";

    public static void main(String[] args) throws IOException, TimeoutException {
        //建立一個連線工廠
        ConnectionFactory factory = new ConnectionFactory();
        //設定RabbitMQ服務的IP、賬號、密碼、Vhost虛擬主機(預設 "/" 則不需要設定)
        factory.setHost("192.168.31.51");
        factory.setUsername("admin");
        factory.setPassword("123");
        factory.setVirtualHost("test");
        //通過工廠物件獲取一個連線
        Connection connection = factory.newConnection();
        //通過連線來獲取一個通道
        Channel channel = connection.createChannel();
        //宣告一個佇列
        //引數一：佇列名稱
        //引數二：佇列裡的訊息是否持久化，預設訊息儲存在記憶體中，預設false
        //引數三：該佇列是否只供一個消費者進行消費的獨佔佇列，則為 true（僅限於此連線），false（預設，可以多個消費者消費）
        //引數四：是否自動刪除 最後一個消費者斷開連線以後 該佇列是否自動刪除 true 自動刪除，預設false
        //引數五：構建佇列的其它屬性，看下面擴充套件引數
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //傳送的訊息
        byte[] msg = "這是一個簡單訊息".getBytes(StandardCharsets.UTF_8);
        //傳送訊息
        //引數一：將傳送到RabbitMQ的哪個交換機上
        //引數二：路由的key是什麼（直接交換機找到路由後，通過路由key來確定最終的佇列）
        //引數三：其它引數
        //引數四：傳送到佇列的具體資訊
        channel.basicPublish("", QUEUE_NAME, null, msg);
        System.out.println("訊息傳送完成！");
    }
}

package cn.xw.helloWorld;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 * @author AnHui OuYang
 * @version 1.0
 * created at 2022-03-05 15:12
 */
public class Consumer {
    //簡單佇列名稱
    public static final String QUEUE_NAME = "helloWorldQueue";

    public static void main(String[] args) throws IOException, TimeoutException {
        //建立一個連線工廠
        ConnectionFactory factory = new ConnectionFactory();
        //設定RabbitMQ服務的IP、賬號、密碼、Vhost虛擬主機(預設 "/" 則不需要設定)
        factory.setHost("192.168.31.51");
        factory.setUsername("admin");
        factory.setPassword("123");
        factory.setVirtualHost("test");
        //通過工廠物件獲取一個連線
        Connection connection = factory.newConnection();
        //通過連線來獲取一個通道
        Channel channel = connection.createChannel();
        System.out.println("消費者開始監聽佇列訊息....");
        //推播的訊息如何進行消費的介面回撥
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                System.out.println("獲取佇列資訊：" + new String(message.getBody(), StandardCharsets.UTF_8));
            }
        };
        //取消消費的一個回撥介面 如在消費的時候佇列被刪除掉了
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
                System.out.println("監聽的佇列出現異常；可能佇列被刪除！");
            }
        };
        //消費者消費訊息
        //引數一：消費哪個佇列
        //引數二：消費成功之後是否要自動應答 true 代表自動應答 false 手動應答
        //引數三：接受佇列訊息的回撥介面
        //引數四：取消消費的回撥介面
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    }
}

x-dead-letter-exchange：
    死信交換器
x-dead-letter-routing-key：
    死信訊息的可選路由鍵
x-expires：
    佇列在指定毫秒數後被刪除
x-message-ttl：
    毫秒為單位的訊息過期時間，佇列級別
x-ha-policy：
    建立HA佇列，此引數已失效
x-ha-nodes：
    HA佇列的分佈節點，此引數已失效
x-max-length：
    佇列的訊息條數限制。限制加入queue中訊息的條數。先進先出原則，超過後，後面的訊息會頂替前面的訊息。
x-max-length-bytes：
    訊息容量限制，該引數和x-max-length目的一樣限制佇列的容量，但是這個是靠佇列大小（bytes）來達到限制。
x-max-priority：
    最大優先值為255的佇列優先排序功能
x-overflow：
    設定佇列溢位行為。這決定了當達到佇列的最大長度時訊息會發生什麼。
    有效值是drop-head、reject-publish或reject-publish-dlx。
x-single-active-consumer：
    表示佇列是否是單一活動消費者，true時，註冊的消費組內只有一個消費者消費訊息，
    其他被忽略，false時訊息迴圈分發給所有消費者(預設false)
x-queue-mode：
    將佇列設定為延遲模式，在磁碟上保留儘可能多的訊息，以減少RAM的使用；
    如果未設定，佇列將保留記憶體快取以儘可能快地傳遞訊息
x-queue-master-locator：
    在叢集模式下設定映象佇列的主節點資訊

public class ProducerA {
    //工作佇列名稱
    public static final String QUEUE_NAME = "workQueue";

    public static void main(String[] args) throws IOException {
        //呼叫自己的工具類獲取通道
        Channel channel = ChannelUtil.getChannel();
        //建立佇列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //迴圈傳送訊息
        for (int i = 0; i < 1000; i++) {
            byte[] msg = ("這是一個編號為：" + i + " 的待處理的訊息").getBytes(StandardCharsets.UTF_8);
            channel.basicPublish("", QUEUE_NAME, null, msg);
        }
        System.out.println("訊息傳送完成！");
    }
}

public class ConsumerA {
    //工作佇列名稱
    public static final String QUEUE_NAME = "workQueue";

    public static void main(String[] args) throws IOException {
        //呼叫自己的工具類獲取通道
        Channel channel = ChannelUtil.getChannel();
        //建立佇列 以防啟動消費者發現佇列不存在報錯
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        System.out.println("消費者A開始監聽佇列訊息....");
        //消費者消費訊息
        channel.basicConsume(QUEUE_NAME, true, (consumerTag, message) -> {
            System.out.println("A消費者獲取佇列資訊並處理：" + new String(message.getBody(), StandardCharsets.UTF_8));
        }, consumerTag -> {
            System.out.println("監聽的佇列出現異常；可能佇列被刪除！");
        });
    }
}

public class ConsumerB {
    //工作佇列名稱
    public static final String QUEUE_NAME = "workQueue";

    public static void main(String[] args) throws IOException {
        //呼叫自己的工具類獲取通道
        Channel channel = ChannelUtil.getChannel();
        //建立佇列 以防啟動消費者發現佇列不存在報錯
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        System.out.println("消費者B開始監聽佇列訊息....");
        //消費者消費訊息
        channel.basicConsume(QUEUE_NAME, true, (consumerTag, message) -> {
            System.out.println("B消費者獲取佇列資訊並處理：" + new String(message.getBody(), StandardCharsets.UTF_8));
        }, consumerTag -> {
            System.out.println("監聽的佇列出現異常；可能佇列被刪除！");
        });
    }
}

注：basicConsume訊息接收方法中的autoAck引數必須為false才可以顯示為手動確認
手動應答分為三種情況:
②：手動拒絕 basicReject(long deliveryTag, boolean requeue)：
    拒絕deliveryTag對應的訊息，第二個引數是否requeue，true則重新入佇列，否則丟棄或者進入死信佇列。
③：手動不確認 basicNack(long deliveryTag, boolean multiple, boolean requeue)
    不確認deliveryTag對應的訊息，第二個引數是否應用於多訊息，第三個引數是否requeue，與basic.reject區別就是同時支援多個訊息，
    可以nack該消費者先前接收未ack的所有訊息。nack後的訊息也會被自己消費到。
③：手動恢復 basicRecover(boolean requeue)
    是否恢復訊息到佇列，true則重新入佇列，並且儘可能的將之前recover的訊息投遞給其他消費者消費，而不是自己再次消費。
    false則訊息會重新被投遞給自己。
④：手動應答 basicAck(long deliveryTag, boolean multiple)
    如果消費者在處理訊息的過程中，出了錯，就沒有什麼辦法重新處理這條訊息，所以在平時都是處理訊息成功後，再確認訊息；
    當autoAck=false時，RabbitMQ會等待消費者手動發回ack訊號後,才從記憶體（和磁碟，如果是持久化訊息的話）中移除訊息。
    它採用訊息確認機制，消費者就有足夠的時間處理訊息(任務)，不用擔心處理訊息過程中消費者程序掛掉後訊息丟失的問題，
    因為RabbitMQ會一直持有訊息直到消費者手動呼叫channel.basicAck為止。對於RabbitMQ伺服器端而言，如果伺服器端一直沒
    有收到消費者的ack訊號，並且消費此訊息的消費者已經斷開連線，則伺服器端會安排該訊息重新進入佇列，等待投遞給下一個
    消費者（也可能還是原來的那個消費者）。這裡我們啟動了手動確認後，就必須呼叫channel.basicAck方法進行確認，
    否則的話RabbitMQ會一直進行等待，當我們這個消費者關閉後，RabbitMQ會將該條訊息再發給對應的消費者進行消費，
    直到有消費者對該條訊息進行消費並應答完成。
引數說明：
    deliveryTag：對應訊息的ID；通過message.getEnvelope().getDeliveryTag()獲取
    requeue：是否重新入列，true代表拒絕應答後會重新返回佇列，false則直接刪除或者進入死信佇列
    multiple：是否批次應答，true代表批次應答
        假設有個佇列依次排列為 1、2、3...10 (1最先出隊，10最後出隊)；
        當為true，傳送1~5訊息給消費者處理完都未確認，當到第6時執行應答方法，並且multiple為true，則代表1~6都被被批次應答
        當為false，傳送1~5訊息給消費者處理完都未確認，當到第6時執行應答方法，並且multiple為true，則代表只要6被應答

//生產者只管發任務訊息，程式碼不變，消費者程式碼優化更改以下，多個消費者程式碼也和這一樣
public class ConsumerB {
    //工作佇列名稱
    public static final String QUEUE_NAME = "workQueue";

    public static void main(String[] args) throws IOException {
        //呼叫自己的工具類獲取通道
        Channel channel = ChannelUtil.getChannel();
        //建立佇列 以防啟動消費者發現佇列不存在報錯
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        System.out.println("消費者B開始監聽佇列訊息....");
        //應答方式 true自動應答  false手動應答(若是手動應答必須設定false)
        boolean autoAck = false;
        //消費者消費訊息requeue
        channel.basicConsume(QUEUE_NAME, autoAck, (consumerTag, message) -> {
            try {
                //這裡我就一句列印語句，沒有複雜邏輯，正常這裡有複雜業務
                System.out.println("B消費者獲取佇列資訊並處理：" + new String(message.getBody(), StandardCharsets.UTF_8));
                int i = 1/0;
                //手動確認應答 不批次應答
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            } catch (IOException e) {
                e.printStackTrace();
                //出現異常手動進行不應答;並且放入佇列中（reject或者使用uack方式都可以，或者本次訊息不處理了可以通過recover重新放到佇列）
                channel.basicReject(message.getEnvelope().getDeliveryTag(), true);
            }
        }, consumerTag -> {
            System.out.println("監聽的佇列出現異常；可能佇列被刪除！");
        });
    }
}

與交換機的持久化相同，佇列的持久化也是通過durable引數實現的(設定後佇列也會有個D)，看一下方法的定義：
queueDeclare(String queue,boolean durable,boolean exclusive,boolean autoDelete,Map<String, Object> arguments)
boolean durable：
    引數跟交換機方法的引數一樣，true表示做持久化，當RabbitMQ服務重啟時，佇列依然存在
boolean exclusive（補充）：
    排它佇列。如果一個佇列被宣告為排他佇列，那麼這個佇列只能被第一次宣告它的連線所見，並在連線斷開的時候自動刪除。
    這裡有三點需要說明：
        1：排它佇列是基於連線可見的，同一連線的不同通道是可以同時存取同一連線建立的排它佇列
        2：如果一個連線已經宣告了一個排它佇列，其它連線是不允許建立同名的排它佇列的，這個與普通佇列不同
        3：即使該佇列是持久化的，一旦連線關閉或者使用者端退出，該排它佇列都會被自動刪除的，這種佇列適用於一
        個使用者端傳送讀取訊息的應用場景
boolean autoDelete（補充）：
    自動刪除，如果該佇列沒有任何訂閱的消費者的話，該佇列會被自動刪除。這種佇列適用於臨時佇列

basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
BasicProperties props設定訊息持久化方式：
引數實現類：
public static class BasicProperties extends com.rabbitmq.client.impl.AMQBasicProperties {
        private String contentType;　　　　　　       //  訊息的內容型別，如：text/plain
        private String contentEncoding;             //  訊息內容編碼
        private Map<String,Object> headers;         //  設定訊息的header,型別為Map<String,Object>
        private Integer deliveryMode;               //  1（nopersistent）非持久化，2（persistent）持久化
        private Integer priority;                   //  訊息的優先順序
        private String correlationId;               //  關聯ID
        private String replyTo;                     //  用於指定回覆的佇列的名稱
        private String expiration;                  //  訊息的失效時間
        private String messageId;                   //  訊息ID
        private Date timestamp;                     //  訊息的時間戳
        private String type;                        //  型別
        private String userId;                      //  使用者ID
        private String appId;                       //  應用程式ID
        private String clusterId;                   //  叢集ID
}
deliveryMode是設定訊息持久化的引數，等於1不設定持久化，等於2設定持久化；
我們平時不會使用BasicProperties類而是使用MessageProperties，通過這個類來獲取具體設定
設定 MessageProperties.PERSISTENT_TEXT_PLAIN
代表：
public static final BasicProperties PERSISTENT_TEXT_PLAIN =
    new BasicProperties("text/plain",null,null,2,0, null, null, null,null, null, null, null,null, null);

//也可以通過這種方式設定；傳送訊息的引數設定 expiration過期時間   deliveryMode 訊息持久化方式
AMQP.BasicProperties properties = new AMQP.BasicProperties()
        .builder().expiration("10000").deliveryMode(2).build();

public class ConsumerA {
    //工作佇列名稱
    public static final String QUEUE_NAME = "workQueue";

    public static void main(String[] args) throws IOException {
        //呼叫自己的工具類獲取通道
        Channel channel = ChannelUtil.getChannel();
        //建立佇列 以防啟動消費者發現佇列不存在報錯
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //設定0代表輪詢分發、1不公平分發、大於1代表預取值
        channel.basicQos(1);
        System.out.println("消費者A(處理資源很快)開始監聽佇列訊息....");
        //應答方式 true自動應答  false手動應答(若是手動應答必須設定false)
        boolean autoAck = false;
        //消費者消費訊息requeue
        channel.basicConsume(QUEUE_NAME, autoAck, (consumerTag, message) -> {
            try {
                //這裡我就一句列印語句，沒有複雜邏輯，正常這裡有複雜業務
                System.out.println("A消費者獲取佇列資訊並處理：" + new String(message.getBody(), StandardCharsets.UTF_8));
                Thread.sleep(3000);    //3秒才能處理完一個任務訊息
                //手動確認應答 不批次應答
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
                //出現異常手動進行不應答;並且放入佇列中
                channel.basicReject(message.getEnvelope().getDeliveryTag(), true);
            }
        }, consumerTag -> {
            System.out.println("監聽的佇列出現異常；可能佇列被刪除！");
        });
    }
}
//消費者A(處理資源很快)開始監聽佇列訊息....
//A消費者獲取佇列資訊並處理：這是一個編號為：0 的待處理的訊息
//A消費者獲取佇列資訊並處理：這是一個編號為：2 的待處理的訊息
//A消費者獲取佇列資訊並處理：這是一個編號為：3 的待處理的訊息
//A消費者獲取佇列資訊並處理：這是一個編號為：4 的待處理的訊息
//A消費者獲取佇列資訊並處理：這是一個編號為：6 的待處理的訊息
//A消費者獲取佇列資訊並處理：這是一個編號為：7 的待處理的訊息
//A消費者獲取佇列資訊並處理：這是一個編號為：8 的待處理的訊息

public class ConsumerB {
    //工作佇列名稱
    public static final String QUEUE_NAME = "workQueue";

    public static void main(String[] args) throws IOException {
        //呼叫自己的工具類獲取通道
        Channel channel = ChannelUtil.getChannel();
        //建立佇列 以防啟動消費者發現佇列不存在報錯
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //設定0代表輪詢分發、1不公平分發、大於1代表預取值
        channel.basicQos(1);
        System.out.println("消費者B(處理資源很慢)開始監聽佇列訊息....");
        //應答方式 true自動應答  false手動應答(若是手動應答必須設定false)
        boolean autoAck = false;
        //消費者消費訊息requeue
        channel.basicConsume(QUEUE_NAME, autoAck, (consumerTag, message) -> {
            try {
                //這裡我就一句列印語句，沒有複雜邏輯，正常這裡有複雜業務
                System.out.println("B消費者獲取佇列資訊並處理：" + new String(message.getBody(), StandardCharsets.UTF_8));
                Thread.sleep(10000);    //10秒才能處理完一個任務訊息
                //手動確認應答 不批次應答
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
                //出現異常手動進行不應答;並且放入佇列中
                channel.basicReject(message.getEnvelope().getDeliveryTag(), true);
            }
        }, consumerTag -> {
            System.out.println("監聽的佇列出現異常；可能佇列被刪除！");
        });
    }
}
//消費者B(處理資源很慢)開始監聽佇列訊息....
//B消費者獲取佇列資訊並處理：這是一個編號為：1 的待處理的訊息
//B消費者獲取佇列資訊並處理：這是一個編號為：5 的待處理的訊息
//B消費者獲取佇列資訊並處理：這是一個編號為：9 的待處理的訊息

public class ProducerA {
    //單個釋出確認
    public static final String SINGLE_RELEASE_CONFIRMATION = "singleReleaseConfirmation";
public static void main(String[] args) throws IOException, InterruptedException {
        long begin = System.currentTimeMillis();    //記錄開始時間
        //獲取通道
        Channel channel = ChannelUtil.getChannel();
        //建立一個通道
        channel.queueDeclare(SINGLE_RELEASE_CONFIRMATION, true, false, false, null);
        //開啟發布確認功能
        channel.confirmSelect();
        //迴圈傳送訊息
        for (int i = 0; i < 1000; i++) {
            String str = "單個釋出確認資訊" + i;
            System.out.println("開始傳送資訊：" + i);
            //釋出資訊
            channel.basicPublish("", SINGLE_RELEASE_CONFIRMATION, MessageProperties.PERSISTENT_TEXT_PLAIN,
                    str.getBytes(StandardCharsets.UTF_8));
            //驗證是否傳送成功（等待確認）
            //channel.waitForConfirms(3000); 傳送三秒後沒得到回覆將斷定未傳送過去
            boolean b = channel.waitForConfirms();
            if (b) {
                System.out.println("傳送成功了：" + i);
            }
        }
        long end = System.currentTimeMillis();  //記錄結尾時間
        System.out.println("單個釋出確認用時：" + (end - begin)); //單個釋出確認用時：2278
    }
}

public class ProducerA {
    //批次確認
    public static final String BATCH_CONFIRMATION = "batchConfirmation";public static void main(String[] args) throws IOException, InterruptedException {
        long begin = System.currentTimeMillis();    //記錄開始時間
        //獲取通道
        Channel channel = ChannelUtil.getChannel();
        //建立一個通道
        channel.queueDeclare(BATCH_CONFIRMATION, true, false, false, null);
        //開啟發布確認功能
        channel.confirmSelect();
        //定義每次批次處理多少訊息進行確認
        int batchNumber = 100;
        //迴圈傳送訊息
        for (int i = 0; i < 1000; i++) {
            String str = "批次釋出確認資訊" + i;
            System.out.println("開始傳送資訊：" + i);
            //釋出資訊
            channel.basicPublish("", BATCH_CONFIRMATION, MessageProperties.PERSISTENT_TEXT_PLAIN,
                    str.getBytes(StandardCharsets.UTF_8));
            //驗證是否傳送成功（等待確認）  用求餘的方式來判斷每輪100個
            //channel.waitForConfirms(3000); 傳送三秒後沒得到回覆將斷定未傳送過去
            if ((i + 1) % batchNumber == 0) {
                if (channel.waitForConfirms()) {
                    System.out.println("批次傳送成功了 範圍為：" + (i - (batchNumber - 1)) + " ~ " + i);
                }
            }
        }
        long end = System.currentTimeMillis();  //記錄結尾時間
        System.out.println("批次釋出確認用時：" + (end - begin)); //批次釋出確認用時：454
    }
}

public class ProducerA {
    //非同步釋出確認
    public static final String ASYNC_RELEASE_CONFIRMATION = "asyncReleaseConfirmation";

    public static void main(String[] args) throws IOException {
        long begin = System.currentTimeMillis();    //記錄開始時間
        //獲取通道
        Channel channel = ChannelUtil.getChannel();
        //建立一個通道
        channel.queueDeclare(ASYNC_RELEASE_CONFIRMATION, true, false, false, null);
        //開啟發布確認功能
        channel.confirmSelect();
        //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Start
        //執行緒安全有序的一個雜湊表Map，適用於高並行的情況
        //1.輕鬆的將序號與訊息進行關聯 2.輕鬆批次刪除條目 只要給到序列號 3.支援並行存取
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
        
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            //這個是回撥成功的，回撥成功後把集合中的資料刪除，最終就代表失敗的多少
            if (multiple) {
                ConcurrentNavigableMap<Long, String> longStringConcurrentNavigableMap =
                        outstandingConfirms.headMap(deliveryTag, true);
                longStringConcurrentNavigableMap.clear();
            } else {
                outstandingConfirms.remove(deliveryTag);
            }
            System.out.println("~~~~ 回撥成功的資料：" + deliveryTag + "   是否批次確認：" + multiple);
        };
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            System.out.println("~~~~ 回撥失敗的資料：" + deliveryTag);
        };
        //新增監聽器，監聽返回(監聽器一定要再傳送訊息之前就建立和監聽) 引數1：回撥成功 引數2：回撥失敗
        channel.addConfirmListener(ackCallback, nackCallback);
        //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ End
        //迴圈傳送訊息 （因為是非同步 這裡正常傳送不用管它）
        for (int i = 0; i < 10000; i++) {
            String str = "非同步釋出確認資訊" + i;
            //記錄要傳送的資料新增到集合中
            outstandingConfirms.put(channel.getNextPublishSeqNo(),str);
            //釋出資訊
            channel.basicPublish("", ASYNC_RELEASE_CONFIRMATION, MessageProperties.PERSISTENT_TEXT_PLAIN,
                    str.getBytes(StandardCharsets.UTF_8));
        }
        long end = System.currentTimeMillis();  //記錄結尾時間
        System.out.println("非同步釋出確認用時：" + (end - begin)); //非同步釋出確認用時：337
    }
}

我們沒使用Exchange，但仍能夠將訊息傳送到佇列。之前能實現的原因是因為我們使用的是預設交換，我們通過空字串("")進行標識
　　basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
　　當初使用：channel.basicPublish("", "當初寫佇列名稱", null,"傳送的訊息");
第一個引數是交換機的名稱：空字串表示預設或無名稱交換機
　　訊息能路由傳送到佇列中其實是由routingKey(binding key)繫結key指定的，那時key都填寫佇列名稱，所有直接被繫結到對應佇列，
　　可以說使用的是直接交換機（direct）

建立交換機方法：exchangeDeclare(String exchange,BuiltinExchangeType type,boolean durable,
                    boolean autoDelete,boolean internal,Map<String, Object> arguments) 
exchange: 交換機名稱
type: 交換機型別，direct、topic、 fanout、 headers
durable: 是否需要持久化
autoDelete: 當最後一個繫結到Exchange上的佇列刪除後，自動刪除該Exchange
internal: 當前Exchange是否用於RabbitMQ內部使用，預設為False
arguments: 擴充套件引數，用於擴充套件AMQP協定客製化化使用
注：推薦在編寫生產者時建立交換機，在編寫消費者時應該建立佇列，並且佇列繫結交換機，啟動時先啟動交換機

上圖是一個佇列繫結關係圖，我們來看看他們之間資料接收情況是怎麼樣的：
     佇列繫結交換機的Key               匹配規則
    quick.orange.rabbit         被佇列 Q1 Q2 接收到
    lazy.orange.elephant        被佇列 Q1 Q2 接收到
    quick.orange.fox            被佇列 Q1 接收到
    lazy.brown.fox              被佇列 Q2 接收到
    lazy.pink.rabbit            雖然滿足兩個繫結規則但兩個規則都是在Q2佇列，所有隻有Q2接收一次
    quick.brown.fox             不匹配任何繫結不會被任何佇列接收到會被丟棄
    quick.orange.male.rabbit    是四個單詞不匹配任何繫結會被丟棄
    lazy.orange.male.rabbit     是四個單詞但匹配 Q2

①：訊息被拒絕（basic.reject/basic.nack)，且 requeue = false(代表不重新回到佇列)
②：訊息因TTL過期（就是任務訊息上攜帶過期時間）
③：訊息佇列的訊息數量已經超過最大佇列長度，先入隊的訊息會被丟棄變為死信

public class TTLConsumer {
    //宣告普通的交換機名稱
    public static final String NORMAL_EXCHANGE = "NormalExchange";
    //宣告死信交換機名稱
    public static final String DLX_EXCHANGE = "DLXExchange";
    //宣告普通佇列名稱
    public static final String Normal_Queue = "NormalQueue";
    //宣告死信佇列名稱
    public static final String DLX_QUEUE = "DLXQueue";
    //宣告路由繫結關係 Routing Key 普通交換機到普通佇列
    public static final String NORMAL_KEY = "NormalKey";
    //宣告路由繫結關係 Routing Key 死信交換機到死信佇列
    public static final String DLX_KEY = "DLXKey";

    public static void main(String[] args) throws IOException {
        //呼叫自己的工具類獲取通道
        Channel channel = ChannelUtil.getChannel();
        //宣告普通exchange交換機 並設定為直接交換機；防止消費者先啟動報錯，找不到交換機
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT,
                true, false, false, null);
        //宣告死信exchange交換機 並設定為直接交換機；防止消費者先啟動報錯，找不到交換機
        channel.exchangeDeclare(DLX_EXCHANGE, BuiltinExchangeType.DIRECT,
                true, false, false, null);

        //宣告死信佇列
        channel.queueDeclare(DLX_QUEUE, true, false, false, null);
        //死信佇列繫結死信交換機routingKey
        channel.queueBind(DLX_QUEUE, DLX_EXCHANGE, DLX_KEY);

        //引數設定
        Map<String, Object> arguments = new HashMap<>();
        //正常佇列設定死信交換機 引數key是固定值
        arguments.put("x-dead-letter-exchange", DLX_EXCHANGE);
        //正常佇列設定死信交換機到死信佇列繫結Routing Key 引數key是固定值
        arguments.put("x-dead-letter-routing-key", DLX_KEY);
        //宣告普通佇列
        channel.queueDeclare(Normal_Queue, true, false, false, arguments);
        //普通佇列繫結普通交換機routingKey
        channel.queueBind(Normal_Queue, NORMAL_EXCHANGE, NORMAL_KEY);
        System.out.println("初始化完成，等待接收訊息");
        //接收佇列訊息
        channel.basicConsume(Normal_Queue, true, (consumerTag, message) -> {
            System.out.println("如同佇列獲取的任務並處理：" +
                    new String(message.getBody(), StandardCharsets.UTF_8));
        }, consumerTag -> {
            System.out.println("監聽的佇列出現異常；可能佇列被刪除！");
        });
    }
}

public class DLXConsumer {
    //宣告死信交換機名稱
    public static final String DLX_EXCHANGE = "DLXExchange";
    //宣告死信佇列名稱
    public static final String DLX_QUEUE = "DLXQueue";
    //宣告路由繫結關係 Routing Key 死信交換機到死信佇列
    public static final String DLX_KEY = "DLXKey";

    public static void main(String[] args) throws IOException {
        //呼叫自己的工具類獲取通道
        Channel channel = ChannelUtil.getChannel();
        //宣告死信exchange交換機 並設定為直接交換機；防止消費者先啟動報錯，找不到交換機
        channel.exchangeDeclare(DLX_EXCHANGE, BuiltinExchangeType.DIRECT,
                true, false, false, null);
        //宣告死信佇列
        channel.queueDeclare(DLX_QUEUE, true, false, false, null);
        //死信佇列繫結死信交換機routingKey
        channel.queueBind(DLX_QUEUE, DLX_EXCHANGE, DLX_KEY);
        System.out.println("初始化完成，等待接收訊息");
        //接收佇列訊息
        channel.basicConsume(DLX_QUEUE, true, (consumerTag, message) -> {
            System.out.println("死信佇列裡獲取的任務並處理：" +
                    new String(message.getBody(), StandardCharsets.UTF_8));
        }, consumerTag -> {
            System.out.println("監聽的佇列出現異常；可能佇列被刪除！");
        });
    }
}

public class DLXProducer {
    //宣告普通的交換機名稱
    public static final String NORMAL_EXCHANGE = "NormalExchange";
    //宣告路由繫結關係 Routing Key 普通交換機到普通佇列
    public static final String NORMAL_KEY = "NormalKey";

    public static void main(String[] args) throws IOException {
        //呼叫自己的工具類獲取通道
        Channel channel = ChannelUtil.getChannel();
        //宣告普通exchange交換機 並設定為直接交換機；防止消費者先啟動報錯，找不到交換機
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT,
                true, false, false, null);
        //傳送訊息的引數設定 expiration過期時間10秒   deliveryMode 訊息持久化方式
        AMQP.BasicProperties properties = new AMQP.BasicProperties()
                .builder().expiration("10000").deliveryMode(2).build();
        //迴圈傳送訊息
        for (int i = 0; i < 5; i++) {
            String str = "測試佇列任務 " + i;
            //釋出資訊
            channel.basicPublish(NORMAL_EXCHANGE, NORMAL_KEY, properties,
                    str.getBytes(StandardCharsets.UTF_8));
        }
        System.out.println("訊息傳送完畢！");
    }
}

 　　　　//引數設定
        Map<String, Object> arguments = new HashMap<>();
        //正常佇列設定死信交換機 引數key是固定值
        arguments.put("x-dead-letter-exchange", DLX_EXCHANGE);
        //正常佇列設定死信交換機到死信佇列繫結Routing Key 引數key是固定值
        arguments.put("x-dead-letter-routing-key", DLX_KEY);
        //設定正常佇列的長度限制 為3 
        arguments.put("x-max-length",3);

//接收佇列訊息
channel.basicConsume(Normal_Queue, false, (consumerTag, message) -> {
    //獲取的任務訊息
    String msg = new String(message.getBody(), StandardCharsets.UTF_8);
    //手動不確認，拒收，並丟去佇列
    if ("測試佇列任務 3".equals(msg)) {
        //出現異常手動進行不應答;並且不放入佇列中
        channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
    } else {
        System.out.println("如同佇列獲取的任務並處理：" + msg);
    }
}, consumerTag -> {
    System.out.println("監聽的佇列出現異常；可能佇列被刪除！");
});

那麼什麼時候需要用延時佇列呢？考慮一下以下場景：
    ①：訂單在十分鐘之內未支付則自動取消。
    ②：新建立的店鋪，如果在十天內都沒有上傳過商品，則自動傳送訊息提醒。
    ③：賬單在一週內未支付，則自動結算。
    ④：使用者註冊成功後，如果三天內沒有登陸則進行簡訊提醒。
    ⑤：使用者發起退款，如果三天內沒有得到處理則通知相關運營人員。
    ⑥：預定會議後，需要在預定的時間點前十分鐘通知各個與會人員參加會議。

設定這個TTL值有兩種方式（佇列設定、訊息設定）：
第一種是在建立佇列的時候設定佇列的 "x-message-ttl" 屬性，如下：
    Map<String, Object> arguments = new HashMap<>();
    //設定訊息延遲10秒；投遞到該佇列的訊息超過10秒直接丟棄
    arguments.put("x-message-ttl",10000);
    //建立佇列，並指定引數
    channel.queueDeclare(Normal_Queue, true, false, false, arguments);
第二種方式針對每條訊息設定TTL：
    AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
    //這條訊息的過期時間也被設定成了10s , 超過10秒未處理則執行到此訊息後被丟棄
    builder.expiration("10000");
    AMQP.BasicProperties properties = builder.build();
    channel.basicPublish(exchangeName, routingKey, mandatory, properties, "msg body".getBytes());

區別的：第一種在佇列上設定TTL屬性，那麼一旦訊息過期，就會被佇列丟棄；而第二種方式，訊息即使過期，也不一定會被馬上丟棄，
因為訊息是否過期是在即將投遞到消費者之前判定的，如果當前佇列有嚴重的訊息積壓情況，則已過期的訊息也許還能存活較長時間。
另外，還需要注意的一點是，如果不設定TTL，表示訊息永遠不會過期，如果將TTL設定為0，則表示除非此時可以直接投遞該訊息到消費者，
否則該訊息將會被丟棄。