RabbitMQ后续

RabbitMQ后续   RabbitMQ高级特性

消息的可靠投递

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。

confirm 确认模式

return 退回模式

rabbitmq 整个消息投递的路径为：

producer—>rabbitmq broker—>exchange—>queue—>consumer

l消息从 producer 到 exchange 则会返回一个 /confirm/iCallback 。

l消息从 exchange–>queue 投递失败则会返回一个 returnCallback 。

我们将利用这两个 callback 控制消息的可靠性投递

提供者代码实现

① 创建项目 rabbitmq-producer-spring

② 添加pom文件

 
        
            org.springframework
            spring-context
            5.1.7.RELEASE
        

        
            org.springframework.amqp
            spring-rabbit
            2.1.8.RELEASE
        

        
            junit
            junit
            4.12
        

        
            org.springframework
            spring-test
            5.1.7.RELEASE
        
    

    
        
            
                org.apache.maven.plugins
                maven-compiler-plugin
                3.8.0
                
                    1.8
                    1.8
                
            
        
    

③ 在resource 文件夹下面添加 配置文件 rabbitmq.properties

rabbitmq.host=192.168.200.142
rabbitmq.port=5672
rabbitmq.username=admin
rabbitmq.password=123456
rabbitmq.virtual-host=/

④ 在resource 文件夹下面添加 配置文件 spring-rabbitmq-producer.xml

⑤ 创建测试类 ， 添加确认模式

package com.Zh;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;


@RunWith(SpringRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {
    @Autowired
     private RabbitTemplate rabbitTemplate;

    @Test
    public void testConfirm() {
        //生产者往交换机发消息,可以得到broker的确认,回调产生的/confirm/i方法,收没有收到都会执行
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {

            
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (ack) {
                    System.out.println("ack==true,交换机收到消息,已经确认了");
                } else {
                    System.out.println("ack=false,交换机没有收到消息  cause="+cause);
                }
            }
        });

        rabbitTemplate.convertAndSend("test_exchange_/confirm/i", "/confirm/i", "test direct exchange");

    }
}

⑥ 添加回退模式

交换机不能把消息投递到队列就会执行这个回调函数,但是一定要有

rabbitTemplate.setMandatory(true);

   @Test
    public void testReturn() {


        //消息必须到达,否则执行回调函数,不加的话回调函数收不到消息,正常发送就不执行回调函数
        rabbitTemplate.setMandatory(true);
        //交换机发送消息到队列,如果找不到对应的队列,就执行returnedMessage
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {

                System.out.println("message = " + message);//发送的消息
                System.out.println("replyCode = " + replyCode);//状态码
                System.out.println("replyText = " + replyText);//为什么
                System.out.println("exchange = " + exchange);//发送的交换机
                System.out.println("routingKey = " + routingKey);//发送的路由key
            }
        });
        //修改一下路由key,改成没有的路由
        rabbitTemplate.convertAndSend("test_exchange_/confirm/i", "confir11m", "test direct exchange 3333");
    }

Consumer Ack

ack指Acknowledge，确认。 表示消费端收到消息后的确认方式。

有二种确认方式：

自动确认：acknowledge=“none” 默认手动确认：acknowledge=“manual”

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。

如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

① 创建项目 rabbitmq-consumer-spring

② 添加pom文件

        
            org.springframework
            spring-context
            5.1.7.RELEASE
        

        
            org.springframework.amqp
            spring-rabbit
            2.1.8.RELEASE
        

        
            junit
            junit
            4.12
        

        
            org.springframework
            spring-test
            5.1.7.RELEASE
        
    

    
        
            
                org.apache.maven.plugins
                maven-compiler-plugin
                3.8.0
                
                    1.8
                    1.8
                
            
        
    

③ 在 resource 文件夹下面新建 rabbitmq.properties 文件 和 spring-rabbitmq-consumer.xml 文件

rabbitmq.properties 文件

rabbitmq.host=192.168.200.142
rabbitmq.port=5672
rabbitmq.username=admin
rabbitmq.password=123456
rabbitmq.virtual-host=/

spring-rabbitmq-consumer.xml 文件

  手动确认

出现异常后消息没有被消费,重回队列

④ 添加监听器

@Component
public class ackListener implements ChannelAwareMessageListener {
    
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {

        long deliveryTag = message.getMessageProperties().getDeliveryTag();//消息的标识,类似数据id

        try {
            System.out.println(new String(message.getBody(),"utf-8"));
            int i = 1 / 0;//出现异常
            channel.basicAck(deliveryTag,true);//true直接确认多条消息,false表示确认一条消息
        } catch (Exception e) {
            e.printStackTrace();
            
            channel.basicNack(deliveryTag,true ,true );
        }

    }


}

 ⑤ 添加测试类

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-consumer.xml")
public class ConsumerTest {
    @Test
    public void test(){
        while (true){

        }
    }
}

运行测试类 ，会一直监听消息 ，查看后台

 当程序报错，程序会拒绝签收，直到修改错误，修改上面的监听器，注释 除 0 错误 ，重新运行程序

Consumer Ack 小结

在rabbit:listener-container标签中设置acknowledge属性，设置ack方式 none：自动确认，manual：手动确认

如果在消费端没有出现异常，则调用channel.basicAck(deliveryTag，true);方法确认签收消息

如果出现异常，则在catch中调用 basicNack，拒绝消息，让MQ重新发送消息。

消费端限流

① 在项目 rabbitmq-consumer-spring ，新建 QosListener

package com.Zh.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;


@Component
public class QosListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        //1.获取消息
        System.out.println(new String(message.getBody()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);

    }
}

 spring-rabbitmq-consumer.xml

rabbitmq-producer-spring

  //测试消费限流
    @Test
    public void testPrefetch() {
        for (int i = 1; i <= 10; i++) {
            rabbitTemplate.convertAndSend("test_exchange_/confirm/i", "/confirm/i", "test direct exchange "+i);
        }
    }

运行程序

⑦ 消费端限流小结

在 中配置 prefetch 属性设置消费端一次拉取多少条消息消费端的必须确认才会继续处理其他消息。

 TTL 

TTL 全称 Time To Live（存活时间/过期时间）。

当消息到达存活时间后，还没有被消费，会被自动清除。

RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。

 控制后台演示消息过期

① 修改管理后台界面，增加队列

参数：表示过期时间，单位毫秒 ，10000表示10秒

 ② 增加交换机

 ③ 绑定队列

④ 发送消息

 ⑤ 查看消息，可以看到消息，但十秒之后，消息自动消失，因为我们设置了十秒消息过期

  代码实现

 队列统一过期

修改 rabbitmq-producer-spring 项目的 配置文件 spring-rabbitmq-producer.xml

在测试类 ProducerTest 中,添加测试方法,发送消息

@Test
public void testTTL() {
     for (int i = 0; i < 10; i++) {
       rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl");
     }
}

查看控制台,发现有10条消息,十秒之后自动过期

 消息过期

rabbitmq-producer-spring

   
    @Test
    public void testMessageTtl() {
        // 消息后处理对象,设置一些消息的参数信息
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {

            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //1.设置message的信息
                // 第二个方法：消息的过期时间 ,5秒之后过期
                message.getMessageProperties().setExpiration("5000");
                //2.返回该消息
                return message;
            }
        };

        //消息单独过期
        rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl....",messagePostProcessor);
    }

死信队列

死信队列，英文缩写：DLX 。DeadLetter Exchange（死信交换机），当消息成为Dead message后，可以被重新发送到另一个交换机，这个交换机就是DLX。

什么是死信队列

先从概念解释上搞清楚这个定义，死信，顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说，producer将消息投递到broker或者直接到queue里了，consumer从queue取出消息进行消费，但某些时候由于特定的原因导致queue中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信，自然就有了死信队列；

消息成为死信的三种情况：

队列消息数量到达限制；比如队列最大只能存储10条消息，而发了11条消息，根据先进先出，最先发的消息会进入死信队列。消费者拒接消费消息，basicNack/basicReject，并且不把消息重新放入原目标队列，requeue=false；原队列存在消息过期设置，消息到达超时时间未被消费；

死信的处理方式

死信的产生既然不可避免，那么就需要从实际的业务角度和场景出发，对这些死信进行后续的处理，常见的处理方式大致有下面几种，

① 丢弃，如果不是很重要，可以选择丢弃

② 记录死信入库，然后做后续的业务分析或处理

③ 通过死信队列，由负责监听死信的应用程序进行处理

综合来看，更常用的做法是第三种，即通过死信队列，将产生的死信通过程序的配置路由到指定的死信队列，然后应用监听死信队列，对接收到的死信做后续的处理，

队列绑定死信交换机：

给队列设置参数： x-dead-letter-exchange 和 x-dead-letter-routing-key

 过期时间代码实现

rabbitmq-producer-spring

修改生产者项目的配置文件 spring-rabbitmq-producer.xml ，增加如下代码

在测试类中，添加如下方法，进行测试

     
    @Test
    public void testDlx(){
        //1. 测试过期时间，死信消息
        rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗？");
    }

运行测试,查看管理台界面

 长度限制代码实现

修改测试类,添加测试方法

  
    @Test
    public void testDlx(){
        //1. 测试过期时间,死信消息
        //rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗？");

        //2. 测试长度限制后,消息死信
        for (int i = 0; i < 20; i++) {
            rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗？");
        }
    }

运行测试方法,进行测试

 测试消息拒收

在消费者工程rabbitmq-consumer-spring 创建DlxListener

@Component
public class DlxListener implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            //1.接收转换消息
            System.out.println(new String(message.getBody()));

            //2. 处理业务逻辑
            System.out.println("处理业务逻辑...");
            int i = 3/0;//出现错误
            //3. 手动签收
            channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            //e.printStackTrace();
            System.out.println("出现异常,拒绝接受");
            //4.拒绝签收,不重回队列 requeue=false
            channel.basicNack(deliveryTag,true,false);
        }
    }
}

 修改消费者配置文件 spring-rabbitmq-consumer.xml

运行消费者测试类

修改生产者测试代码

  
    @Test
    public void testDlx(){
        //1. 测试过期时间,死信消息
        //rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗？");

        //2. 测试长度限制后,消息死信
//        for (int i = 0; i < 20; i++) {
//            rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗？");
//        }
        //3. 测试消息拒收
        rabbitTemplate.convertAndSend("test_exchange_dlx",
                                     "test.dlx.haha",
                                         "我是一条消息,我会死吗？");
    }

发送消息,运行程序,查看后台管理界面

死信队列小结

死信交换机和死信队列和普通的没有区别当消息成为死信后，如果该队列绑定了死信交换机，则消息会被死信交换机重新路由到死信队列消息成为死信的三种情况：

队列消息长度（数量）到达限制；消费者拒接消费消息，并且不重回队列；原队列存在消息过期设置，消息到达超时时间未被消费；

延迟队列

延迟队列存储的对象肯定是对应的延时消息，所谓”延时消息”是指当消息被发送以后，并不想让消费者立即拿到消息，而是等待指定时间后，消费者才拿到这个消息进行消费。

场景：在订单系统中，一个用户下单之后通常有30分钟的时间进行支付，如果30分钟之内没有支付成功，那么这个订单将进行取消处理。这时就可以使用延时队列将订单信息发送到延时队列。

需求：

下单后，30分钟未支付，取消订单，回滚库存。新用户注册成功30分钟后，发送短信问候。

实现方式：

延迟队列

 

很可惜，在RabbitMQ中并未提供延迟队列功能。

但是可以使用：TTL+死信队列 组合实现延迟队列的效果。

 代码实现

在rabbitmq-producer-spring

修改生产者代码 ，修改生产者配置文件 spring-rabbitmq-producer.xml

修改生产者，添加测试方法 

 @Test
    public  void testDelay() throws InterruptedException {
        //1.发送订单消息。 将来是在订单系统中,下单成功后,发送消息
        rabbitTemplate.convertAndSend("order_exchange",
                "order.msg","订单信息：id=1,time=2020年10月17日11：41：47");

        //2.打印倒计时10秒
        for (int i = 10; i > 0 ; i--) {
            System.out.println(i+"...");
            Thread.sleep(1000);
        }
    }

运行程序创建订单延时队列

 消费者

修改消费者项目,添加OrderListener

消费者

package com.Zh.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

@Component
public class OrderListener implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            //1.接收转换消息
            System.out.println(new String(message.getBody()));

            //2. 处理业务逻辑
            System.out.println("处理业务逻辑...");
            System.out.println("根据订单id查询其状态...");
            System.out.println("判断状态是否为支付成功");
            System.out.println("取消订单,回滚库存....");
            //3. 手动签收
            channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            //e.printStackTrace();
            System.out.println("出现异常,拒绝接受");
            //4.拒绝签收,不重回队列 requeue=false
            channel.basicNack(deliveryTag,true,false);
        }
    }
}

修改消费者配置文件 spring-rabbitmq-consumer.xml

    

        

        

        

        

        

        

    

运行消费者测试类