【RabbitMQ】一文带你搞定springboot整合RabbitMQ涉及消息的发送确认,消息的消费确认机制,延时队列的实现

摘要:
新创建的店铺,如果在十天内都没有上传过商品,则自动发送消息提醒。账单在一周内未支付,则自动结算。用户发起退款,如果三天内没有得到处理则通知相关运营人员。预定会议后,需要在预定的时间点前十分钟通知各个与会人员参加会议。
说明

这一篇里,我们将继续介绍RabbitMQ的高级特性,通过本篇的学习,你将收获:

  • 什么是延时队列
  • 延时队列使用场景
  • RabbitMQ中的TTL
  • 如何利用RabbitMQ来实现延时队列
本文大纲

【RabbitMQ】一文带你搞定springboot整合RabbitMQ涉及消息的发送确认,消息的消费确认机制,延时队列的实现第1张

什么是延迟队列

延时队列,首先,它是一种队列,队列意味着内部的元素是有序的,元素出队和入队是有方向性的,元素从一端进入,从另一端取出。

其次,延时队列,最重要的特性就体现在它的延时属性上,跟普通的队列不一样的是,普通队列中的元素总是等着希望被早点取出处理,而延时队列中的元素则是希望被在指定时间得到取出和处理,所以延时队列中的元素是都是带时间属性的,通常来说是需要被处理的消息或者任务。

简单来说,延时队列就是用来存放需要在指定时间被处理的元素的队列

延迟队里的使用场景

那么什么时候需要用延时队列呢?考虑一下以下场景:

  • 订单在十分钟之内未支付则自动取消。
  • 新创建的店铺,如果在十天内都没有上传过商品,则自动发送消息提醒。
  • 账单在一周内未支付,则自动结算。
  • 用户注册成功后,如果三天内没有登陆则进行短信提醒。
  • 用户发起退款,如果三天内没有得到处理则通知相关运营人员。
  • 预定会议后,需要在预定的时间点前十分钟通知各个与会人员参加会议。
    这些场景都有一个特点,需要在某个事件发生之后或者之前的指定时间点完成某一项任务,如:发生订单生成事件,在十分钟之后检查该订单支付状态,然后将未支付的订单进行关闭;发生店铺创建事件,十天后检查该店铺上新商品数,然后通知上新数为0的商户;发生账单生成事件,检查账单支付状态,然后自动结算未支付的账单;发生新用户注册事件,三天后检查新注册用户的活动数据,然后通知没有任何活动记录的用户;发生退款事件,在三天之后检查该订单是否已被处理,如仍未被处理,则发送消息给相关运营人员;发生预定会议事件,判断离会议开始是否只有十分钟了,如果是,则通知各个与会人员。

看起来似乎使用定时任务,一直轮询数据,每秒查一次,取出需要被处理的数据,然后处理不就完事了吗?如果数据量比较少,确实可以这样做,比如:对于“如果账单一周内未支付则进行自动结算”这样的需求,如果对于时间不是严格限制,而是宽松意义上的一周,那么每天晚上跑个定时任务检查一下所有未支付的账单,确实也是一个可行的方案。但对于数据量比较大,并且时效性较强的场景,如:“订单十分钟内未支付则关闭“,短期内未支付的订单数据可能会有很多,活动期间甚至会达到百万甚至千万级别,对这么庞大的数据量仍旧使用轮询的方式显然是不可取的,很可能在一秒内无法完成所有订单的检查,同时会给数据库带来很大压力,无法满足业务要求而且性能低下。

更重要的一点是,不!优!雅!

没错,作为一名有追求的程序员,始终应该追求更优雅的架构和更优雅的代码风格,写代码要像写诗一样优美。【滑稽】

这时候,延时队列就可以闪亮登场了,以上场景,正是延时队列的用武之地。

既然延时队列可以解决很多特定场景下,带时间属性的任务需求,那么如何构造一个延时队列呢?接下来,本文将介绍如何用RabbitMQ来实现延时队列。

RabbitMq中的TTL

在介绍延时队列之前,还需要先介绍一下RabbitMQ中的一个高级特性——TTL(Time To Live)。

TTL是什么呢?TTL是RabbitMQ中一个消息或者队列的属性,表明一条消息或者该队列中的所有消息的最大存活时间,单位是毫秒。换句话说,如果一条消息设置了TTL属性或者进入了设置TTL属性的队列,那么这条消息如果在TTL设置的时间内没有被消费,则会成为“死信”(至于什么是死信,请翻看上一篇)。如果同时配置了队列的TTL和消息的TTL,那么较小的那个值将会被使用。
那么,如何设置这个TTL值呢?有两种方式,第一种是在创建队列的时候设置队列的“x-message-ttl”属性,如下:

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-message-ttl", 6000);
channel.queueDeclare(queueName, durable, exclusive, autoDelete, args);

这样所有被投递到该队列的消息都最多不会存活超过6s。
另一种方式便是针对每条消息设置TTL,代码如下:

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
builder.expiration("6000");
AMQP.BasicProperties properties = builder.build();
channel.basicPublish(exchangeName, routingKey, mandatory, properties, "msg body".getBytes());

这样这条消息的过期时间也被设置成了6s。

但这两种方式是有区别的,如果设置了队列的TTL属性,那么一旦消息过期,就会被队列丢弃,而第二种方式,消息即使过期,也不一定会被马上丢弃,因为消息是否过期是在即将投递到消费者之前判定的,如果当前队列有严重的消息积压情况,则已过期的消息也许还能存活较长时间。

另外,还需要注意的一点是,如果不设置TTL,表示消息永远不会过期,如果将TTL设置为0,则表示除非此时可以直接投递该消息到消费者,否则该消息将会被丢弃。

思路:rabbitMQ 如何实现

1、rabbitMQ为每个队列设置消息的超时时间。只要给队列设置x-message-ttl 参数,就设定了该队列所有消息的存活时间,时间单位是毫秒。如果声明队列时指定了死信交换器,则过期消息会成为死信消息
2、需要设置的参数为:
【RabbitMQ】一文带你搞定springboot整合RabbitMQ涉及消息的发送确认,消息的消费确认机制,延时队列的实现第2张

原理:上图

【RabbitMQ】一文带你搞定springboot整合RabbitMQ涉及消息的发送确认,消息的消费确认机制,延时队列的实现第3张
1、将延迟队列(queue)在声明的时候设置参数 “ x-dead-letter-exchange ”,“ x-message-ttl “ 分别对应 死信路由器(dlx_exchange) 和 消息过期时间(比如说30分钟)。
2、一个消息从生产者发送到延迟队列 ,在延迟队列里等待,等待30分钟后,会去绑定的死信路由(dlx_exchange)。通过死信路由的规则,走到死信队列。
3、这时候监听死信队列的消费者就可以接收到消息,消费消息。比如说查看该订单是否支付,如果没有支付,则关闭该订单。

实战演练

springboot整合RabbitMQ涉及消息的发送确认,消息的消费确认机制

1.引入maven依赖
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2.在application.yml的配置:
spring:
  rabbitmq:
    host: 106.52.82.241
    port: 5672
    username: yang
    password: Yangxiaohui227
    virtual-host: /
    publisher-confirms: true #消息发送后,如果发送成功到队列,则会回调成功信息
    publisher-returns: true  #消息发送后,如果发送失败,则会返回失败信息信息
    listener:  #加了2下面2个属性,消费消息的时候,就必须发送ack确认,不然消息永远还在队列中
      direct:
        acknowledge-mode: manual
      simple:
        acknowledge-mode: manual
//为了统一管理所有的Mq消息,建一个类存储常量,消息的设计都基本会涉及(队列(queue),交换机(exchange),路由键(route)三个值)
public class RabbitMqConstant {

    //下单发送消息 队列名,交换机名,路由键的配置
    public final static String SHOP_ORDER_CREATE_EXCHANGE="shop.order.create.exchange";
    public final static String SHOP_ORDER_CREATE_ROUTE="shop.order.create.route";
    public final static String SHOP_ORDER_CREATE_QUEUE="shop.order.create.queue";
}
package com.example.demo.mq;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
//该类是mq最重要的一个类,所有队列的创建,交换机的创建,队列和交换机的绑定都在这里实现
@Configuration
public class RabbitMqConfig {
    private final static Logger log = LoggerFactory.getLogger(RabbitMqConfig.class);
    @Autowired
    private CachingConnectionFactory connectionFactory;

    @Autowired
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    /**
     * 单一消费者
     *
     * @return
     */

    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        factory.setTxSize(1);
        return factory;
    }

    /**
     * 多个消费者
     *
     * @return
     */
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory, connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(20);
        factory.setMaxConcurrentConsumers(20);
        factory.setPrefetchCount(20);
        return factory;
    }

    /**
     * 模板的初始化配置
     *
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean sucess, String cause) {
                if (sucess) {
                    log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, sucess, cause);
                }

            }
        });
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.warn("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message);
            }
        });
        return rabbitTemplate;
    }



    //消息的创建设计三个步骤:队列的创建,交换机创建(direct类型,topic类型,fanout类型),队列和交换机的通过路由键的绑定


    //--------- 下单消息配置
    //队列
    @Bean
    public Queue shopOrderCreateQueue() {
        return new Queue(RabbitMqConstant.SHOP_ORDER_CREATE_QUEUE, true);
    }

    //Direct交换机(一对一关系,一个direct交换机只能绑定一个队列,当有2个相同消费者时,如项目部署2台机,只有一个消费者能消费,)
    @Bean
    DirectExchange shopOrderCreateExchange() {
        return new DirectExchange(RabbitMqConstant.SHOP_ORDER_CREATE_EXCHANGE);
    }

    //绑定
    @Bean
    Binding bindShopOrderCreateQueue() {
        return BindingBuilder.bind(shopOrderCreateQueue()).to(shopOrderCreateExchange()).with(RabbitMqConstant.SHOP_ORDER_CREATE_ROUTE);
    }
}
import com.alibaba.fastjson.JSON;
import com.example.demo.domain.ShopOrderMast;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
//专门用一个类作为消息的生产者
@Service
public class ShopMessagePublisher {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendCreateOrderMessage(ShopOrderMast orderMast){
        CorrelationData correlationData=new CorrelationData(); //该参数可以传,可以不传,不传时,correlationData的id值默认是null,消息发送成功后,在RabbitMqConfig类的rabbitTemplate类的confirm方法会接收到该值
        correlationData.setId(orderMast.getCodOrderId()); 
        String msg = JSON.toJSONString(orderMast);
        //convertAndSend该方法有非常多的重构方法,找到适合自己的业务方法就行了,这里我用的是其中一个,发送时指定exchange和route值,这样就会发到对应的队列去了
        rabbitTemplate.convertAndSend(RabbitMqConstant.SHOP_ORDER_CREATE_EXCHANGE,RabbitMqConstant.SHOP_ORDER_CREATE_ROUTE,msg,correlationData);

    }
}
//所有的消费都写在一个消费类中
@Service
public class ShopMessageComsumer {
    //监听下单消息
    @RabbitListener(queues =RabbitMqConstant.SHOP_ORDER_CREATE_QUEUE)
    public void createOrderMesaageComsumer(String msg, Channel channel, Message message) {
        try {
                //消息可以通过msg获取也可以通过message的body属性获取
                System.out.println("开始消费了");
                ShopOrderMast shopOrderMast = JSON.parseObject(msg, ShopOrderMast.class);


            /**
             * 因为我在application.yml那里配置了消息手工确认也就是传说中的ack,所以消息消费后必须发送确认给mq
             * 很多人不理解ack(消息消费确认),以为这个确认是告诉消息发送者的,这个是错的,这个ack是告诉mq服务器,
             * 消息已经被我消费了,你可以删除它了
             * 如果没有发送basicAck的后果是:每次重启服务,你都会接收到该消息
             * 如果你不想用确认机制,就去掉application.yml的acknowledge-mode: manual配置,该配置默认
             * 是自动确认auto,去掉后,下面的channel.basicAck就不用写了
             *
             */
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

            } catch (Exception e) {
            try {
                //出现异常,告诉mq抛弃该消息
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
                e.printStackTrace();
            } catch (IOException e1) {
                e1.printStackTrace();
            }

        }

    }
}
//这里我发送了一条消息,orderId我设置为555556666666,在消息发送时,存到了CorrelationData对象中,因此,发送成功后,在confirm方法可以接收到该值了
//消息发送成功后,在控制台会看到有成功的回调信息,也就是回调了rabbitTemplate的:
confirm(CorrelationData correlationData, boolean sucess, String cause)

【RabbitMQ】一文带你搞定springboot整合RabbitMQ涉及消息的发送确认,消息的消费确认机制,延时队列的实现第4张
//上面测试的下单消息是direct类型消息的,现在创建一个topic消息

//RabbitMqConstant新增topic的配置信息
//下单topic消息:路由键的名字 星号* 代表多个字符,#号代表一个字符
    //topic交换机,发送消息时,发送到指定shop.order.create.topic.exchange和shop.order.create.topic.route中
    public final static String SHOP_ORDER_CREATE_TOPIC_EXCHANGE="shop.order.create.topic.exchange";
    public final static String SHOP_ORDER_CREATE_TOPIC_TOUTE="shop.order.create.topic.route";


    //队列1,通过shop.order.create.topic.*与交换机绑定
    public final static String SHOP_ORDER_CREATE_TOPIC_ROUTE_ONE="shop.order.create.topic.*";
    public final static String SHOP_ORDER_CREATE_TOPIC_QUEUE_ONE="shop.order.create.topic.queue.one";


    //队列2 通过shop.order.create.topic.*与交换机绑定shop.order.create.topic.#
    public final static String SHOP_ORDER_CREATE_TOPIC_ROUTE_TWO="shop.order.create.topic.#";
    public final static String SHOP_ORDER_CREATE_TOPIC_QUEUE_TWO="shop.order.create.topic.queue.two";
//在RabbitMqConfig新增topic队列的基本信息
//-------------------------下单TOPIC消息的创建

    //创建TOPIC交换机
    @Bean
    TopicExchange shopOrderCreateTopicExchange() {
        return new TopicExchange(RabbitMqConstant.SHOP_ORDER_CREATE_TOPIC_EXCHANGE);
    }
    //---------------------------//队列1使用自己的route和交换机绑定
    //创建队列1
    @Bean
    public Queue shopOrderCreateQueueOne() {
        return new Queue(RabbitMqConstant.SHOP_ORDER_CREATE_TOPIC_QUEUE_ONE, true);
    }
    //绑定
    @Bean
    Binding bindShopOrderCreateQueueOne() {
        return BindingBuilder.bind(shopOrderCreateQueueOne()).to(shopOrderCreateTopicExchange()).with(RabbitMqConstant.SHOP_ORDER_CREATE_TOPIC_ROUTE_ONE);
    }

    //---------------------------//队列2用自己的route和交换机绑定

    //创建队列2
    @Bean
    public Queue shopOrderCreateQueueTWO() {
        return new Queue(RabbitMqConstant.SHOP_ORDER_CREATE_TOPIC_QUEUE_TWO, true);
    }

    //绑定
    @Bean
    Binding bindShopOrderCreateQueueTWO() {
        return BindingBuilder.bind(shopOrderCreateQueueTWO()).to(shopOrderCreateTopicExchange()).with(RabbitMqConstant.SHOP_ORDER_CREATE_TOPIC_ROUTE_TWO);
    }

//消息的发送方新增
  //发送TOPIC消息
    public void sendCreateOrderTOPICMessage(ShopOrderMast orderMast){
        CorrelationData correlationData=new CorrelationData(); //该参数可以传,可以不传,不传时,correlationData的id值默认是null,消息发送成功后,在RabbitMqConfig类的rabbitTemplate类的confirm方法会接收到该值
        correlationData.setId(orderMast.getCodOrderId());
        String msg = JSON.toJSONString(orderMast);
        //消息发送使用公共route而不是某个队列自己的route
        rabbitTemplate.convertAndSend(RabbitMqConstant.SHOP_ORDER_CREATE_TOPIC_EXCHANGE,RabbitMqConstant.SHOP_ORDER_CREATE_TOPIC_TOUTE,msg,correlationData);

    }
//消息的消费方新增
//消费者1
    @RabbitListener(queues =RabbitMqConstant.SHOP_ORDER_CREATE_TOPIC_QUEUE_ONE)
    public void createOrderMesaageComsumerOne(String msg, Channel channel, Message message) {
        try {
                //消息可以通过msg获取也可以通过message对象的body值获取
                System.out.println("我是消费者1");
                ShopOrderMast shopOrderMast = JSON.parseObject(msg, ShopOrderMast.class);


            /**
             * 因为我在application.yml那里配置了消息手工确认也就是传说中的ack,所以消息消费后必须发送确认给mq
             * 很多人不理解ack(消息消费确认),以为这个确认是告诉消息发送者的,这个是错的,这个ack是告诉mq服务器,
             * 消息已经被我消费了,你可以删除它了
             * 如果没有发送basicAck的后果是:每次重启服务,你都会接收到该消息
             * 如果你不想用确认机制,就去掉application.yml的acknowledge-mode: manual配置,该配置默认
             * 是自动确认auto,去掉后,下面的channel.basicAck就不用写了
             *
             */
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

            } catch (Exception e) {
            try {
                //出现异常,告诉mq抛弃该消息
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
                e.printStackTrace();
            } catch (IOException e1) {
                e1.printStackTrace();
            }

        }

    }
    //消费者2
    @RabbitListener(queues =RabbitMqConstant.SHOP_ORDER_CREATE_TOPIC_QUEUE_TWO)
    public void createOrderMesaageComsumerTWO(String msg, Channel channel, Message message) {
        try {
                //消息可以通过msg获取也可以通过message对象的body值获取
                System.out.println("我是消费者2");
                ShopOrderMast shopOrderMast = JSON.parseObject(msg, ShopOrderMast.class);


            /**
             * 因为我在application.yml那里配置了消息手工确认也就是传说中的ack,所以消息消费后必须发送确认给mq
             * 很多人不理解ack(消息消费确认),以为这个确认是告诉消息发送者的,这个是错的,这个ack是告诉mq服务器,
             * 消息已经被我消费了,你可以删除它了
             * 如果没有发送basicAck的后果是:每次重启服务,你都会接收到该消息
             * 如果你不想用确认机制,就去掉application.yml的acknowledge-mode: manual配置,该配置默认
             * 是自动确认auto,去掉后,下面的channel.basicAck就不用写了
             *
             */
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

            } catch (Exception e) {
            try {
                //出现异常,告诉mq抛弃该消息
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
                e.printStackTrace();
            } catch (IOException e1) {
                e1.printStackTrace();
            }

        }

    }

//测试结果:
【RabbitMQ】一文带你搞定springboot整合RabbitMQ涉及消息的发送确认,消息的消费确认机制,延时队列的实现第5张

springboot整合RabbitMQ延时队列的实现

延时队列:将消息发送到一个队列,等过了一段时间后,该队列会将消息转发到真正的队列消费,业务场景可以用于订单定时取消

//在RabbitMqConstant类添加如下内容
 //延时队列,消息先发到延时队列中,到时间后,再发送到真正的队列

    public final static String SHOP_ORDER_CREATE_DELAY_EXCHANGE="shop.order.create.delay.exchange";
    public final static String SHOP_ORDER_CREATE_DELAY_ROUTE="shop.order.create.delay.route";
    public final static String SHOP_ORDER_CREATE_DELAY_QUEUE="shop.order.create.delay.queue";

    //真正的队列

    public final static String SHOP_ORDER_CREATE_REAL_EXCHANGE="shop.order.create.real.exchange";
    public final static String SHOP_ORDER_CREATE_REAL_ROUTE="shop.order.create.real.route";
    public final static String SHOP_ORDER_CREATE_REAL_QUEUE="shop.order.create.real.queue";
//在RabbitMqConfig加上
 //----------------------- 延时队列的配置

    //延时队列
    @Bean
    public Queue shopOrderCreateDelayQueue() {
        Map<String, Object> argsMap= Maps.newHashMap();
        argsMap.put("x-dead-letter-exchange",RabbitMqConstant.SHOP_ORDER_CREATE_REAL_EXCHANGE); //真正的交换机
        argsMap.put("x-dead-letter-routing-key",RabbitMqConstant.SHOP_ORDER_CREATE_REAL_ROUTE); //真正的路由键
        return new Queue(RabbitMqConstant.SHOP_ORDER_CREATE_DELAY_QUEUE,true,false,false,argsMap);

    }
    //延时交换机
    @Bean
    DirectExchange shopOrderCreateDelayExchange() {
        return new DirectExchange(RabbitMqConstant.SHOP_ORDER_CREATE_DELAY_EXCHANGE);
    }

    //延时队列绑定延时交换机
    @Bean
    Binding bindShopOrderCreateDelayQueue() {
        return BindingBuilder.bind(shopOrderCreateDelayQueue()).to(shopOrderCreateDelayExchange()).with(RabbitMqConstant.SHOP_ORDER_CREATE_DELAY_ROUTE);
    }


    //真正的队列配置-------------------------------------


    //真正的队列
    @Bean
    public Queue shopOrderCreateRealQueue() {

        return new Queue(RabbitMqConstant.SHOP_ORDER_CREATE_REAL_QUEUE,true);

    }
    //真正的交换机
    @Bean
    DirectExchange shopOrderCreateRealExchange() {
        return new DirectExchange(RabbitMqConstant.SHOP_ORDER_CREATE_REAL_EXCHANGE);
    }

    //绑定真正的交换机
    @Bean
    Binding bindShopOrderCreateRealQueue() {
        return BindingBuilder.bind(shopOrderCreateRealQueue()).to(shopOrderCreateRealExchange()).with(RabbitMqConstant.SHOP_ORDER_CREATE_REAL_ROUTE);
    }
//在消息发送类(ShopMessagePublisher)新增
 //发送延时消息
    public void sendCreateOrderDelayMessage(ShopOrderMast orderMast){
        CorrelationData correlationData=new CorrelationData(); //该参数可以传,可以不传,不传时,correlationData的id值默认是null,消息发送成功后,在RabbitMqConfig类的rabbitTemplate类的confirm方法会接收到该值
        correlationData.setId(orderMast.getCodOrderId());
        String msg = JSON.toJSONString(orderMast);
        // convertAndSend(String exchange, String routingKey, Object message, MessagePostProcessor messagePostProcessor, @Nullable CorrelationData correlationData)
        rabbitTemplate.convertAndSend(RabbitMqConstant.SHOP_ORDER_CREATE_DELAY_EXCHANGE, RabbitMqConstant.SHOP_ORDER_CREATE_DELAY_ROUTE, msg, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                MessageProperties messageProperties = message.getMessageProperties();
                messageProperties.setExpiration("60000");//单位是毫秒
                return message;
            }
        }, correlationData);

    }
//在消费类(ShopMessageComsumer) 新增
 //延迟队列中真正队列监听
    @RabbitListener(queues =RabbitMqConstant.SHOP_ORDER_CREATE_REAL_QUEUE)
    public void createOrderRealMesaageComsumer(String msg, Channel channel, Message message) {
        try {

                System.out.println("这是真正的队列,在监听延时队列发送的消息");
                ShopOrderMast shopOrderMast = JSON.parseObject(msg, ShopOrderMast.class);



            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

            } catch (Exception e) {
            try {
                //出现异常,告诉mq抛弃该消息
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
                e.printStackTrace();
            } catch (IOException e1) {
                e1.printStackTrace();
            }

        }

    }

注意,如果同时使用了延时队列的queue去接收,那么消息会被延迟队列的消费者消费,而不是被真正的queue消费

//如果在延迟队列消费时,加了下面这个队列,上面那个真正的消费者就接收不到消息了
    @RabbitListener(queues =RabbitMqConstant.SHOP_ORDER_CREATE_DELAY_QUEUE)
    public void createOrderDelayMesaageComsumer(String msg, Channel channel, Message message) {
        try {
                System.out.println("测试延迟队列自己能否接收");
                ShopOrderMast shopOrderMast = JSON.parseObject(msg, ShopOrderMast.class);
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

            } catch (Exception e) {
            try {
                //出现异常,告诉mq抛弃该消息
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
                e.printStackTrace();
            } catch (IOException e1) {
                e1.printStackTrace();
            }

        }

    }

补充:对于direct和topic交换机,如果部署多台相同queue的消费者,消息也只会消费一次,通过轮询的方式进行负债均衡

如何在rabbitMq管理页面查看没有还没被消费的消息信息:

【RabbitMQ】一文带你搞定springboot整合RabbitMQ涉及消息的发送确认,消息的消费确认机制,延时队列的实现第6张
【RabbitMQ】一文带你搞定springboot整合RabbitMQ涉及消息的发送确认,消息的消费确认机制,延时队列的实现第7张
【RabbitMQ】一文带你搞定springboot整合RabbitMQ涉及消息的发送确认,消息的消费确认机制,延时队列的实现第8张

免责声明:文章转载自《【RabbitMQ】一文带你搞定springboot整合RabbitMQ涉及消息的发送确认,消息的消费确认机制,延时队列的实现》仅用于学习参考。如对内容有疑问,请及时联系本站处理。

上篇常用鼠标事件找到集合 P 中的所有 ”最大“ 点的集合下篇

宿迁高防,2C2G15M,22元/月;香港BGP,2C5G5M,25元/月 雨云优惠码:MjYwNzM=

相关文章

JSR303校验的基本使用和结合AOP使用

[一]:基本使用 springboot默认导入第三方校验框架hibernate-validator 使用JSR303三大步骤 一.先给需要校验数据的javabean上标注校验注解 public class UmsAdminParam { @ApiModelProperty(value = "用户名", required = true) @L...

CentOS 6.3下Samba服务器的安装与配置方法(图文详解)

这篇文章主要介绍了CentOS 6.3下Samba服务器的安装与配置方法(图文详解),需要的朋友可以参考下 一、简介Samba是一个能让Linux系统应用Microsoft网络通讯协议的软件,而SMB是Server Message Block的缩写,即为服务器消息块 ,SMB主要是作为Microsoft的网络通讯协议,后来Samba将SMB通信协议应用到...

kafka 0.8.2 消息消费者 consumer

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven...

springboot 实现注解获取操作日志

1.第一步加入pom依赖 <dependency>   <groupId>org.springframework.boot</groupId>   <artifactId>spring-boot-starter-aop</artifactId> </dependency> 2.第二步加...

Java Servlet 中文API说明

JavaServlet中文API说明 这份文档描述了Java Servlet API的最新版本2.1版。所以,这本书对于Servlet的开发者及servlet引擎的开发者同样适用。 Java Servlet API的组成 Java Servlet API由两个软件包组成:一个是对应HTTP的软件包,另一个是不对应HTTP的通用的软件包。这两个软件包的同时...

权限认证机制

一、Form表单认证 之前的项目以MVC为主,采用的是from表单认证,Forms认证示意图如下: HTTP是一个无状态的协议,WEB服务器在处理所有传入HTTP请求时,根本就不知道某个请求是否是一个用户的第一次请求与后续请求,或者是另一个用户的请求。 WEB服务器每次在处理请求时,都会按照用户所访问的资源所对应的处理代码,从头到尾执行一遍,然后输出响应...