初探RabbitMQ系列（二）深入了解RabbitMQ高级特性

  • Z时代
  • 2024-01-10
  • 分类：综合

编程

上一篇文章：初探RabbitMQ系列（一）了解消息中间件的工作原理和使用，我们详细介绍了RabbitMQ的核心概念、Exchange、Queue、Binding、RoutingKey、Message进行展开讲解，本片内容会对RabbitMQ的高级特性展开说明。

如何保障消息100%投递成功？

生产端的可靠性投递

1.消息的成功发送 -> 2.MQ节点的成功接收 -> 3.发送端收到MQ节点（Broker）确认应答 -> 4.完善消息进行补偿机制

可靠性投递解决方案

  • 消息落库，对消息状态进行打标

  • 消息的延迟投递，做二次确认，回调检查

消息落库

延迟投递

 

 

 

MQ实现消息的幂等性

(一)什么是幂等性？

幂等性的概念：一个幂等方法，无论执行多少次，得到的结果都是一致的，比如：http GET 方法，用于获取资源，不管调用多少次接口，结果都不会改变，所以是幂等的。

比如我们执行一条更新库存的SQL语句（乐观锁）

update T_REPS set count = count - 1,version = version + 1 where version = 1

第一步，查出version，第二步，通过这个version进行更新，这样可实现幂等性保障。

(二)如何避免消息的重复消费？

在海量订单产生的业务高峰期，如何避免消息的重复消费？

  • 解决方法：消息端需要实现幂等性，这就意味着，我们的消息永远不会被消费多次，即使受到多条一样的消息。

业界主流的幂等性决方案主要有两种：

1.唯一ID + 指纹码机制：利用数据库的主键去重

指纹码：为什么需要指纹码呢？主要是用来应对用户在一瞬间的频繁操作，这个指纹码可能是我们外部自定义的一些规则来实现，比如：时间戳+具体的信息码，主要是用来保证这次操作的绝对唯一。

select count(1) from tb_order where ID = 唯一ID + 指纹码

优点：实现简单，通过简单的拼接，然后通过查询判断是否重复。

缺点：在高并发情景下，数据库写入会出现性能瓶颈。

解决方案：根据 ID 进行分库分表，对 id 进行算法路由，落到一个具体的数据库，然后当这个 id 第二次来又会落到这个数据库，这时候就像我单库时的查重一样了。利用算法路由把单库的幂等变成多库的幂等，分摊数据流量压力，提高性能。

2.利用Redis的原子性去实现

我们都知道Redis是单线程的，并且性能也非常好，redis提供了很多原子性操作，比如setnx命令，在接收到消息后将消息ID作为key执行 setnx 命令，如果执行成功就表示没有处理过这条消息，可以进行消费了，执行失败表示消息已经被消费了。

使用 redis 的原子性去实现主要需要考虑两个点：

第一：业务是否需要进行数据落库，如果落库的话，那么关键的解决的问题是数据库和缓存如何做到原子性？

我们都知道redis是单线程的，并且性能也非常好，提供了 很多原子性的命令。比如可以使用 setnx 命令。

 

第二：如果不进行落库，那么都存储到缓存中，如何设置定时同步的策略(同步到关系型数据库)？缓存又如何做到数据可靠性保障呢？

采用延迟双删策略

 

Confirm消息确认机制

消息的确认是指生产者投递消息后，如果Broker收到消息，则会给生产者一个应答；

生产者接收应答，用户来确定这条消息是否正常的发送到了Broker，这种消息确认机制是消息可靠性投递的核心保障。

实现步骤

第一步：在channel上开启确认模式：channel.confirmSelect()

第二步：在channel上添加监听：addConfirmListener，监听成功和失败返回结果，根据具体的结果对消息进程重新发送，或者记录日志等后出处理。

代码实例：

Producer

package com.example.mq.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;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
/**
 * @Author 林必昭
 * @Date 2019/12/1 13:22
 * @descr
 */
public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建ConnectionFactory
        ConnectionFactory factory= new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        //2.获取Connection
        Connection connection = factory.newConnection();
        //3.通过Connection创建一个新的channel
        Channel channel = connection.createChannel();
        //4.指定消息的投递模式：消息的确认模式
        channel.confirmSelect();
        String exchangeName = "test_confirm_name";
        String routingKey = "confirm.save";
        //5.发送一条消息
        String msg = "Hello RabbitMQ Send Confirm Message!";
        channel.basicPublish(exchangeName,routingKey,null,msg.getBytes(StandardCharsets.UTF_8));
        //6.添加一个确认监听
        channel.addConfirmListener(new ConfirmListener() {
            /**
             * 成功响应的处理方法
             * @param deliveryTag   消息投递的唯一标签
             * @param multiple      是否是批量消息
             * @throws IOException
             */
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("--------消费端响应ack!---------");
            }
            /**
             * 失败的处理方法
             */
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("-------------no ack!-------------");
            }
        });
    }
}

Concumer

package com.example.mq.confirm;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * @Author 林必昭
 * @Date 2019/12/1 13:37
 * @descr
 */
public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        //1.创建ConnectionFactory
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        //2.获取Connection
        Connection connection = factory.newConnection();
        //3.通过Connection创建一个新的channel
        Channel channel = connection.createChannel();
        String exchangeName = "test_confirm_name";
        String queueName = "test_queue_name";
        String routingKey = "confirm.#";
        //4.声明交换机和队列 然后进行绑定，最后指定路由键
        channel.exchangeDeclare(exchangeName, "topic", true);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        //5.创建消费者
        while (true) {
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [消费者] 接收到了: "" + message + """);
            };
            channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
            });
        }
    }
}

启动RabbitMQ Service，运行Consumer和Producer，观察控制台打印：

接着启动Producer，发现消费端收到的生产者投递的消息：

观察生产者，返回了消费者的确认消费信息：

  • 通过上述演示，我们完整的实现了Confirm消息确认方案！！！

 

Return返回消息

自定义消费者

消息的ACK与重回队列

消息的限流

设置消息过TTL

死信队列

 

参考文章：http://www.mamicode.com/info-detail-2838117.html

以上是 初探RabbitMQ系列（二）深入了解RabbitMQ高级特性 的全部内容， 来源链接： utcz.com/z/511384.html

回到顶部