RabbitMQ之消息模式

本文轉載自查看原文 2019-12-06 12:11 376 Linux/ SpringCloud

目的：　　

消息如何保證100%的投遞

冪等性概念

Confirm確認消息

Return返回消息

自定義消費者

前言：

　　想必知道消息中間件RabbitMQ的小伙伴，對於引入中間件的好處可以起到抗高並發，削峰，業務解耦的作用並不陌生。

　　康康簡單流程圖了解一下。詳情了解RabbitMQ可移步：https://www.cnblogs.com/huangting/p/11989597.html

注意：一般MQ中間件為了提高系統的吞吐量會把消息保存在內存中，如果不作其他處理，MQ服務器一旦宕機，消息將全部丟失；

　　這樣會造成巨大影響，在業務流程上是不可以的。

以下就是我們去解決所使用的方法

消息如何保證100%的投遞

　　什么是生產端的可靠性投遞？

保障消息的成功發出
保障MQ節點的成功接收
發送端收到MQ節點（Broker）確認應答
完善的消息進行補償機制（如網絡問題沒有返回確認應答）

　　BAT/TMD互聯網大廠的解決方案：

消息落庫，對消息狀態進行打標
消息的延遲投遞，做二次確認，回調檢查

冪等性概念

　　首先，我們了解一下什么叫冪等：

　　　　　　在分布式應用中，冪等是非常重要的，也就是相同條件下對一個業務的操作，不管操作多少次，結果都是一樣。

　　　其次，為什么要有冪等這種場景？

　　　　　　基本上在邏輯涉及范圍比較大系統里邊，都有部署分布式，就如同下面其中訂單和庫存可以說是兩種獨立的個體。

　　　　正因為是分布式，那么很有可能訂單服務在調用庫存時因為網絡差等外因導致調用失敗，但是庫存服務其實是調用成功的只是在返回結果集時發生了異常，其實按照一般思維就會在調用一次庫存不就可以了嗎？恰恰是這樣問題就出現了，訂單沒有接收到返回結果，但庫存已經減去一個對應物品數量，如果在調用一次又減去一次，那么很明顯業務上是錯誤的。

所以這個時候就要用到冪等性，無論庫存服務在相同條件下調用幾次，處理結果都一樣。這樣才能保證業務的緊密。

　業界主流的冪等性操作

樂觀鎖方案

　　比如我們執行一條更新庫存的SQL語句

 Update t_repository set count = count -1,version = version + 1 where version = 1

　　根據version版本，也就是在操作庫存前先獲取當前商品的version版本號，然后操作的時候帶上此version號。

　我們梳理下，我們第一次操作庫存時，得到version為1，調用庫存服務version變成了；

　但返回給訂單服務出現了問題，訂單服務又一次發起調用庫存服務，當訂單服務傳如的version還是1，再執行上面的sql語句時，就不會執行；

　因為version已經變為2了，where條件就不成立。這樣就保證了不管調用幾次，只會真正的處理一次

唯一ID+指紋碼機制，利用數據庫主鍵去重

　　　　原理就是利用數據庫主鍵去重，業務完成后插入主鍵標識

Select count(1) from T_order where ID=唯一ID+指紋碼

唯一ID就是業務表的唯一的主鍵，如商品ID
指紋碼就是為了區別每次正常操作的碼，每次操作時生成指紋碼；可以用時間戳+業務編號的方式。

好處：實現簡單

壞處：高並發下數據庫瓶頸

解決方案：根據ID進行分庫分表進行算法路由

利用Redis的原子性去實現

我們是否需要把業務結果進行數據落庫，如果落庫，關鍵解決的問題時數據庫和redis操作如何做到原子性？

　　　　　　　　意思就是庫存減1了，但redis進行操作完成標記時，失敗了怎么辦？也就是一定要保證落庫和redis 要么一起成功，要么一起失敗　　

如果不進行落庫，那么都存儲到緩存中，如何設置定時同步策略？

　　　　　　意思就是庫存減1，不落庫，直接先操作redis操作完成標記，然后由另外的同步服務進行庫存落庫，這個就是增加了系統復雜性，而且同步策略如何設置

Confirm確認消息

　　　　Confirm消息確認機制含義：

消息的確認，是指訂單服務投遞消息后，如果物流服務收到消息，則會給我們訂單服務一個應答。

中間件進行接收應答，用來確定這條消息是否正常的發送到物流服務，這種方式也是消息的可靠性投遞的核心保障

如何實現Confirm確認消息？

在Channel上開啟確認模式：channel.confirmSelect()
在channel上添加監聽：addConfirmListener，監聽成功和失敗的返回結果，根據具體的結果對消息進行重新發送、或記錄日志等后續處理！

　　消費端代碼

package com.javaxh.rabbitmqapi.confirm;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
public class Consumer {
    public static void main(String[] args) throws Exception {
        //1 創建ConnectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.239.131");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");

        //2 獲取C    onnection
        Connection connection = connectionFactory.newConnection();

        //3 通過Connection創建一個新的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 創建消費者
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
        channel.basicConsume(queueName, true, queueingConsumer);

        while(true){
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());

            System.err.println("消費端: " + msg);
        }


    }
}

服務提供方代碼

package com.javaxh.rabbitmqapi.confirm;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;

public class Producer {
    public static void main(String[] args) throws Exception {
        //1 創建ConnectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.239.131");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");

        //2 獲取C    onnection
        Connection connection = connectionFactory.newConnection();

        //3 通過Connection創建一個新的Channel
        Channel channel = connection.createChannel();

        //4 指定我們的消息投遞模式: 消息的確認模式
        channel.confirmSelect();
        String exchangeName = "test_confirm_exchange";
        String routingKey = "confirm.save";

        //5 發送一條消息
        String msg = "Hello RabbitMQ Send confirm message!";
        channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());

        //6 添加一個確認監聽
        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!-----------");
            }
        });
    }
}

運行結果：

消費端：

服務提供方：

Return返回消息

　　Return Listener用於處理一些不可路由的消息。

正常情況：我們的消息生產者，通過指定一個Exchange和RoutingKey，把消息送達到某一個隊列中去，然后我們的消費者監聽隊列，進行消費處理操作！

異常情況：在某些情況下，如果我們在發送消息的時候，當前的Exchange不存在或者指定的路由key路由不到，這個時候如果我們需要監聽這種不可達的消息，就需要使用Return Listener！

在基礎API中有一個關鍵的配置項

Mandatory：如果為true，則監聽器會接收到路由不可達的消息，然后進行后續處理，如果為false，那么Broker端自動刪除該消息！

消費端代碼

package com.javaxh.rabbitmqapi.returnlistener;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
public class Consumer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.239.131");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");

        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        String exchangeName = "test_return_exchange";
        String routingKey = "return.#";
        String queueName = "test_return_queue";

        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);

        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);

        channel.basicConsume(queueName, true, queueingConsumer);

        while(true){
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.err.println("消費者: " + msg);
        }

    }
}

生產端代碼：

package com.javaxh.rabbitmqapi.returnlistener;
import com.rabbitmq.client.*;

import java.io.IOException;

public class Producer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.239.131");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");

        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        String exchange = "test_return_exchange";
        String routingKey = "return.save";
        String routingKeyError = "abc.save";

        String msg = "Hello RabbitMQ Return Message";


        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange,
                                     String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {

                System.err.println("---------handle  return----------");
                System.err.println("replyCode: " + replyCode);
                System.err.println("replyText: " + replyText);
                System.err.println("exchange: " + exchange);
                System.err.println("routingKey: " + routingKey);
                System.err.println("properties: " + properties);
                System.err.println("body: " + new String(body));
            }
        });

        //消息投遞成功，會被消費者所消費
//        channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
        //消息不可達，將觸發ReturnListener
         channel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());
    }
}

效果:

自定義消費者

我們一般就是在代碼中編寫while循環，進行consumer.nextDelivery方法進行獲取下一條消息，然后進行消費處理！

但是我們使用自定義的Consumer更加的方便，解耦性更加的強，也是實際工作中最常用的使用方式！

自定義消費端代碼

package com.javaxh.rabbitmqapi.consumer;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import java.io.IOException;

public class MyConsumer extends DefaultConsumer {

    public MyConsumer(Channel channel) {
        super(channel);
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.err.println("-----------consume message----------");
        System.err.println("consumerTag: " + consumerTag);
        System.err.println("envelope: " + envelope);
        System.err.println("properties: " + properties);
        System.err.println("body: " + new String(body));
    }
}

消費端調用：

package com.javaxh.rabbitmqapi.consumer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Consumer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.239.131");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");

        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();


        String exchangeName = "test_consumer_exchange";
        String routingKey = "consumer.#";
        String queueName = "test_consumer_queue";

        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);

        channel.basicConsume(queueName, true, new MyConsumer(channel));


    }
}

生產端調用：

package com.javaxh.rabbitmqapi.consumer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Producer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.239.131");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");

        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        String exchange = "test_consumer_exchange";
        String routingKey = "consumer.save";

        String msg = "Hello RabbitMQ Consumer Message";

        for(int i =0; i<5; i ++){
            channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
        }

    }
}

效果：

謝謝觀看！！！

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 RabbitMQ 消息模式 RabbitMQ消息分發模式 RabbitMQ消息交換模式簡介 RabbitMQ基本用法、消息分發模式、消息持久化、廣播模式 RabbitMQ的Topic模式發送與接收消息 RabbitMQ事務性消息和確認模式 SpringBoot---RabbitMQ消息隊列_3(Routing模式、通配符模式) RabbitMQ系列(八)--順序消費模式和迅速消息發送模式 RabbitMQ - 常用消息隊列之：Worker模式（一對多模式） RabbitMQ模式，RabbitMQ和springboot整合，RabbitMQ全鏈路消息不丟失解決