RMQ可靠性投遞

一.什么是RMQ的可靠性投遞

1.保障消息的成功發(fā)出
2.保障MQ節(jié)點的成功接收
3.發(fā)送端收到MQ節(jié)點（broker）確認應答
4.完善的消息補償機制
在實際生產中，很難保障前三點的完全可靠欧募，比如在極端的環(huán)境中，生產者發(fā)送消息失敗了处坪，發(fā)送端在接受確認應答時突然發(fā)生網絡閃斷等等情況，很難保障可靠性投遞颤枪，所以就需要有第四點完善的消息補償機制寿烟。

二虽抄、大廠的解決方案

第一種：消息落庫

消息落庫流程圖.JPG

延遲投遞實現流程.JPG

雖然第二種方案也是無法做到100%的可靠傳遞焰扳，在特別極端的情況倦零，還是需要定時任務和補償機制進行輔助的。但是第二種方案的核心是減少數據庫操作蓝翰，在高并發(fā)場景下光绕，我們考慮的不是百分百的可靠性了，而是考慮可用性畜份，性能能否扛得住這個流量诞帐，所以我能減少一次數據庫操作就減少一次。我上游服務減少了一次數據庫操作爆雹，我的服務性能相對而言就提高了一些停蕉，而且又能把異步callback Server補償服務解耦出來。

confirm機制

消息確認機制.JPG

1. 消息的確認钙态，是指生產投遞消息后慧起，如果broker收到消息，會給生產者一個應答
2. 生產者接受應答册倒，用來確認這條消息是否正常發(fā)送到broker
   如何實現消息的確認機制蚓挤？
   第一步在channe開啟確認模式，channel.confirmSelect();
   第二步在channel添加監(jiān)聽，addConfirmListener,根據監(jiān)聽的結果對消息進行重新發(fā)送或記錄日志等處理灿意。
   代碼如下：
   Producer類

package com.bfxy.rabbitmq.api.confirm;

import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Producer { 
    public static void main(String[] args) throws Exception {   
        //1 創(chuàng)建ConnectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("10.136.198.52");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("123456");    
        //2 獲取C onnection
        Connection connection = connectionFactory.newConnection();  
        //3 通過Connection創(chuàng)建一個新的Channel
        Channel channel = connection.createChannel();       
        //4 指定我們的消息投遞模式: 消息的確認模式 
        channel.confirmSelect();        
        String exchangeName = "test_confirm_exchange";
        String routingKey = "confirm.save"; 
        //5 發(fā)送一條消息
        String msg = "Hello RabbitMQ Send confirm message!";
        channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());       
        //6 添加一個確認監(jiān)聽
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.err.println("-------no ack!-----------");
            }           
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.err.println("-------ack!-----------");
            }
        });     
    }
}

Consumer類

package com.bfxy.rabbitmq.api.confirm;

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

public class Consumer { 
    public static void main(String[] args) throws Exception {               
        //1 創(chuàng)建ConnectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("10.136.198.52");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("123456");        
        //2 獲取C onnection
        Connection connection = connectionFactory.newConnection();      
        //3 通過Connection創(chuàng)建一個新的Channel
        Channel channel = connection.createChannel();       
        String exchangeName = "test_confirm_exchange";
        String routingKey = "confirm.#";
        String queueName = "test_confirm_queue";        
        //4 聲明交換機和隊列 然后進行綁定設置, 最后制定路由Key
        channel.exchangeDeclare(exchangeName,"topic",true);
        channel.queueDeclare(queueName,true,false,false,null);      channel.queueBind(queueName,exchangeName,routingKey);
        //5 創(chuàng)建消費者
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
        //6 設置信道    channel.basicConsume(queueName,true,queueingConsumer);
        //7 消費消息內容
        while (true){
        Delivery delivery = queueingConsumer.nextDelivery();
        String msg = new String(delivery.getBody());
            System.out.println("消費端："+ msg);
        }       
    }
}

先啟動消費者估灿，再啟動生產者: 消費者打印出消費成功的信息，然后重新發(fā)送消息缤剧，生產者接收到確認消費成功的消息后馅袁，打印出確認信息。
首先消費者控制臺如下

comsumer.JPG

producer.JPG