一、訊息佇列學習

MQ是一個網際網路架構中常見的解耦利器。
什麼時候不使用MQ？
上游實時關注執行結果
什麼時候使用MQ？
1）資料驅動的任務依賴，多個任務需要輪流執行，輪流訂閱上一個任務。
2）上游不關心多下游執行結果，上游執行完傳送到MQ，多下游訂閱MQ。
3）非同步返回執行時間長

死信佇列多了一個過期的機制，到期會自動去嘗試消費，通過死信交換機做匹配，選擇想要的消費失敗的訊息。
https://ke.qq.com/course/288116#term_id=100341172 入門視訊
訊息中介軟體是在訊息傳輸過程中儲存訊息的容器。佇列的主要目的是提供路由並保證訊息的傳遞。
特點

：
1非同步處理模式
2多個應用程式呼叫關係為鬆耦合關係
傳遞模型：
1、點多點模型PTP
每個訊息只用一個消費者
傳送者和接收者沒有時間依賴
接受者確認訊息接受和處理成功
2、釋出-訂閱模型Pub/Sub
一對多關係，通過訂閱主題，釋出者建立一個訂閱，訂閱者保持持續的活動狀態以接收訊息。
每個訊息可以有多個訂閱者
客戶端只有訂閱後才能接收到訊息，有時間依賴。
持久訂閱 訂閱關係建立後，訊息不會消失，不管訂閱者是否都線上
非持久訂閱 訂閱者為了接受訊息，必須一直線上

典型的應用案例
1、註冊時傳送郵件或傳送簡訊
2、日誌分析使用，多個服務產生的資料傳送到中介軟體傳送到分析服務。
3、訊息複製，用於跨機房資料傳輸、搜尋、離線資料計算等。
4、延遲訊息傳送和暫存，把中介軟體當成可靠的訊息暫存地。接受訊息，暫時先不傳送。

二、RabbitMQ的安裝

一、安裝erlang，環境變數配置


輸入erl –version檢視安裝是否成功。

二、rabbitMQ的安裝
下載exe，安裝。cd到其sbin目錄。

執行外掛命令rabbitmq-plugins enable rabbitmq_management 開啟管理後臺的外掛

三、JAVA操作佇列

連線工具，配置資訊

1、簡單佇列
此時為匿名傳送，不指定交換機，則直接傳送到佇列中。



2、工作佇列WorkQueue
模型（訊息輪流被多個消費者消費）

此時有兩個消費者同時監聽一個的佇列。
其中一個執行緒sleep2秒，另一個消費者執行緒sleep1秒，但是處理的訊息是一樣多。
這種方式叫輪詢分發（round-robin）

不管誰忙，都不會多給訊息，總是你一個我一個。

想要做到公平分發（fair dispatch），必須關閉自動應答ack，改成手動應答。
使用basicQos(perfetch=1)限制每次只發送不超過1條訊息到同一個消費者，消費者必須手動反饋告知佇列，才會傳送下一個。

ACK（Message acknowledgment）預設是true
autoACK=true時，訊息分發給消費者，則訊息就會從記憶體中刪除。如果正在執行的消費者宕機，則會丟失正在處理的訊息。
所以autoACK=false時，則需要等待訊息應答後才刪除，如果消費者宕機，則會交付給其他消費者。

如果rabbitMQ宕機，則記憶體中的訊息全部丟失。所以需要將訊息持久化。（已經宣告好的佇列不能更改）

注意：RabbitMQ不允許重新定義一個已經存在的佇列。控制檯刪除或者重新定義一個佇列。

durable：持久化，本來宣告在記憶體，持久化會儲存到本地，重啟mq不會消失
Exclusive：排外，如果是，兩個消費者不能同時消費，一般就只有一個消費者
autoDelete：自動刪除，最後一個消費者斷開的時候自動刪除佇列

3、訂閱者模式

一個生產者，多個消費者，消費者都有自己的佇列，訊息先發送到交換機exchange，每個佇列都繫結到交換機。實現一個訊息被多個消費者消費。
佇列如果不繫結到交換機，訊息丟失，交換機沒有儲存能力。
交換機：一方面是接收生產者的訊息，另一方面是向佇列推送訊息。生產者在釋出的時候不指定交換機，則為匿名傳送。交換機型別如下
Fanout（不處理路由鍵），佇列繫結到交換機，直接轉發到所有佇列。
Direct（處理路由鍵）
Topic（）

消費者1和2分設定兩個佇列名稱，繫結到交換機上，可以實現一個訊息兩個消費者訂閱。

4、路由模式
宣告交換機為direct，傳送路由key為error的訊息。
根據繫結的路由key，訊息帶哪個key，就路由到哪個佇列。可以一個佇列繫結多個key

宣告交換機為topic 可以實現模式匹配
#匹配一個或多個 *匹配一個

四、訊息確認機制 事務+confirm

可以通過持久化資料解決rabbitMQ伺服器異常的資料丟失問題。但此時生產者並不知道訊息是否發到rabbitMQ伺服器中。如果生產者要確認訊息已傳送到MQ，可以通過兩種方式。
1、通過AMQP協議方式，實現事務機制
txSelect—txCommit提交事務 txRollback回滾事務
使用者將當前channel設定成transation模式 此時會導致大量的請求，極大的降低了吞吐量

2、生產者端引入confirm模式
佇列如果已宣告為tx不能重新設定為confirm
實現原理

好處：非同步的處理
channel.setConfirmSelect()
程式設計模式
1、普通，發一條會呼叫waitForConfirm()
2、批量，每傳送一批訊息後再呼叫
3、非同步的confirm模式，提供回撥
單條或者批量（序列）傳送confirm的生產者如下：

非同步的confirm 集合存每一個未confirm的訊息，confirm了，就從中刪除

public class Send {
    private static  final  String QUEUE_NAME = "test_queue_confirm2";
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //開啟confirm模式
        channel.confirmSelect();

        //未確認的訊息放入
        SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());

        //新增監聽通道
        channel.addConfirmListener(new ConfirmListener() {
            //沒有問題的handleAck 成功
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                if(multiple){   //多個的
                    System.out.println("handleAck:multiple");
                    confirmSet.headSet(deliveryTag+1).clear();
                }else{          //單個的
                    System.out.println("handleAck:multiple false");
                    confirmSet.remove(deliveryTag);
                }
            }
            //有問題的反饋Nack 失敗
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                if(multiple){
                    System.out.println("handleNack:multiple");
                    confirmSet.headSet(deliveryTag+1).clear();
                }else{
                    System.out.println("handleNack:multiple false");
                    confirmSet.remove(deliveryTag);
                }
            }
        });

        String msg = "hello Confirm11111111";
        while(true){
            long seqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
            confirmSet.add(seqNo);
        }
    }
}

五、SpringBoot整合RabbitMQ

在application.yml做mq配置

宣告佇列，交換機，路由鍵

生產者

消費者