乐趣区

关于rabbitmq:RabbitMQ2-死信队列延迟队列优先队列

jiezi

1. 前言

在《RabbitMQ（1）- 根底开发利用》中，咱们曾经介绍了 RabbitMQ 的根底开发利用。本文基于这些根底再做一些扩大，延长出一些高级的用法，如：死信队列、提早队列和优先队列。不过还是以死信队列为主，因为提早队列是死信队列的衍生概念，而且优先队列也比较简单，所以先还是在代码层面上，把死信队列搞透。

1.1. 创立队列、交换机

咱们在应用 RabbitMQ 之前，须要先创立好相干的队列和交换机，并且设置一些绑定关系。因为几篇文章都是联合 springboot 来开发，上面就联合 springboot 介绍几种创立形式：

  1. 间接拜访 RabbitMQ Management 治理页面，在页面上创立；或者应用 RabbitMQ 其余的客户端来创立治理。
  2. 在 springboot 上基于生产端开发时，@RabbitListener 注解的 bindings属性，能够简略实现相似性能。
@RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "direct.queue.d",
                    durable = "true"),
            exchange = @Exchange(value = "direct.exchange.a",
                    durable = "true",
                    type = ExchangeTypes.DIRECT,
                    ignoreDeclarationExceptions = "true"),
            key = "direct.routingKey.a"
    )
    )
  1. 在配置类下定义 @Bean，即向 Ioc 容器中注册 Queue、Exchange、Binding 的实例。
package pers.kerry.exercise.rabbitmq.rabbitmqproducera.config;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @description:
 * @date: 2020/7/12 11:26 下午
 * @author: kerry
 */
@Configuration
public class RabbitConfig {
   
    public static final String NORMAL_EXCHANGE_A="demo.direct.exchange.a";
    public static final String NORMAL_ROUTING_KEY_A="demo.direct.routingKey.a";
    public static final String NORMAL_QUEUE_A="demo.direct.queue.a";


    /**
     * NORMAL 交换机
     * @return
     */
    @Bean
    public Exchange ExchangeA(){
        return ExchangeBuilder
                .directExchange(NORMAL_EXCHANGE_A)
                .durable(true)
                .build();}

    /**
     * NORMAL 队列
     * @return
     */
    @Bean
    public Queue QueueA(){
        return QueueBuilder
                .durable(NORMAL_QUEUE_A)
                .build();}

    /**
     * 绑定 NORMAL 队列 和 NORMAL 交换机
     * @return
     */
    @Bean
    public Binding normalBinding(){
        return new Binding(NORMAL_QUEUE_A,
                Binding.DestinationType.QUEUE,
                NORMAL_EXCHANGE_A,
                NORMAL_ROUTING_KEY_A,
                null);
    }
}

我集体举荐第三种，而且倡议是在 生产者端 定义，消费者应该更关注生产的逻辑。然而如果用代码来创立，有一个很大的毛病，就是不能删除和批改，至多我目前还没找到方法。

因而要联合第一种和第三种来应用，当然都用第一种也是能够的。只是开发人员，更心愿队列、交换机的创立、绑定的逻辑，都体现在代码外面，通过代码能够更好的浏览架构设计。

2. 死信队列

2.1. 简介

死信队列 这个名字听起来很特地，但它解决的是日常开发中最常见的问题：不能失常生产的音讯，该如何解决。咱们在第一篇文章中有应用到手动 Ack，对于须要 nack 并且无需重回队列的音讯，冀望有对立的异样解决；包含有些音讯是有时效性的，如果领取订单个别都最大领取时常，超时后就应该勾销订单；等等。

死信队列就是应答这些状况的，它呈现的条件如下：

  1. 音讯被否定确认，应用basicNack 或basicReject，并且此时 requeue 属性被设置为 false。
  2. 音讯在队列的存活工夫超过设置的 TTL 工夫。
  3. 音讯队列的音讯数量曾经超过最大队列长度。

死信队列的架构如下：

生产者 --> 音讯 --> 业务交换机 --> 业务队列 --> 音讯变成死信 --> 死信交换机 --> 死信队列 --> 消费者

2.2. 配置

如何配置死信队列呢？其实很简略，大略能够分为以下步骤：

  1. 配置业务队列，绑定到业务交换机上
  2. 为业务队列配置死信交换机和路由 key
  3. 为死信交换机配置死信队列

留神，并不是间接申明一个公共的死信队列，而后所以死信音讯就本人跑到死信队列里去了。而是为每个须要应用死信的业务队列配置一个死信交换机，这里同一个我的项目的死信交换机能够共用一个，而后为每个业务队列调配一个独自的路由 key。

有了死信交换机和路由 key 后，接下来，就像配置业务队列一样，配置死信队列，而后绑定在死信交换机上。也就是说，死信队列并不是什么非凡的队列，只不过是绑定在死信交换机上的队列。死信交换机也不是什么非凡的交换机，只不过是用来承受死信的交换机，所以能够为任何类型【Direct、Fanout、Topic】。一般来说，会为每个业务队列调配一个独有的路由 key，并对应的配置一个死信队列进行监听，也就是说，个别会为每个重要的业务队列配置一个死信队列。

2.3. 代码（生产者）

依照 简介 中死信队列的架构，咱们在配置文件中定义了 NORMAL 的业务队列和业务交换机，以及 DLX 的死信队列和死信交换机，并在业务队列中设置了死信交换机。

RabbitConfig.java

package pers.kerry.exercise.rabbitmq.rabbitmqproducera.config;


import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @description:
 * @date: 2020/7/12 11:26 下午
 * @author: kerry
 */
@Configuration
public class RabbitConfig {
    /**
     * DLX，定义参数
     */
    public static final String X_DEAD_LETTER_EXCHANGE="x-dead-letter-exchange";
    public static final String X_DEAD_LETTER_ROUTING_KEY="x-dead-letter-routing-key";
    public static final String X_MESSAGE_TTL="x-message-ttl";
    public static final String X_MAX_LENGTH="x-max-length";
    /**
     * DLX，命名
     */
    public static final String DEAD_LETTER_EXCHANGE_A="demo.direct.dlx.exchange.a";
    public static final String DEAD_LETTER_ROUTING_KEY_A="demo.direct.dlx.routingKey.a";
    public static final String DEAD_LETTER_QUEUE_A="demo.direct.dlx.queue.a";
    /*
     * NORMAL，命名
     */
    public static final String NORMAL_EXCHANGE_A="demo.direct.exchange.a";
    public static final String NORMAL_ROUTING_KEY_A="demo.direct.routingKey.a";
    public static final String NORMAL_QUEUE_A="demo.direct.queue.a";


    @Bean("jsonRabbitTemplate")
    public RabbitTemplate jsonRabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate=new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate;
    }

    @Bean("defaultRabbitTemplate")
    public RabbitTemplate defaultRabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate=new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    }

    /**
     * DLX 交换机
     * @return
     */
    @Bean
    public Exchange dlxExchangeA(){
        return ExchangeBuilder
                .directExchange(DEAD_LETTER_EXCHANGE_A)
                .durable(true)
                .build();}

    /**
     * DLX 队列
     * @return
     */
    @Bean
    public Queue dlxQueueA(){
        return QueueBuilder
                .durable(DEAD_LETTER_QUEUE_A)
                .build();}

    /**
     * NORMAL 交换机
     * @return
     */
    @Bean
    public Exchange ExchangeA(){
        return ExchangeBuilder
                .directExchange(NORMAL_EXCHANGE_A)
                .durable(true)
                .build();}

    /**
     * NORMAL 队列
     * @return
     */
    @Bean
    public Queue QueueA(){
        return QueueBuilder
                .durable(NORMAL_QUEUE_A)
                // 设置 死信交换机
                .withArgument(X_DEAD_LETTER_EXCHANGE,DEAD_LETTER_EXCHANGE_A)
                .withArgument(X_DEAD_LETTER_ROUTING_KEY,DEAD_LETTER_ROUTING_KEY_A)
                // 设置 队列所有音讯 存活工夫 8 秒
                .withArgument(X_MESSAGE_TTL,8000)
                // 设置 队列最大长度 10 条
                .withArgument(X_MAX_LENGTH,10)
                .build();}

    /**
     * 绑定 DLX 队列 和 DLX 交换机
     * @return
     */
    @Bean
    public Binding dlxBinding(){
        return new Binding(DEAD_LETTER_QUEUE_A,
                Binding.DestinationType.QUEUE,DEAD_LETTER_EXCHANGE_A,
                DEAD_LETTER_ROUTING_KEY_A,
                null);
    }

    /**
     * 绑定 NORMAL 队列 和 NORMAL 交换机
     * @return
     */
    @Bean
    public Binding normalBinding(){
        return new Binding(NORMAL_QUEUE_A,
                Binding.DestinationType.QUEUE,
                NORMAL_EXCHANGE_A,
                NORMAL_ROUTING_KEY_A,
                null);
    }

}

ProducerService.java

@Slf4j
@Service
public class ProducerService {public void sendText(String data, MessageProperties messageProperties) {
        /**
         * 对单个音讯 设置 TTL
         */
        //messageProperties.setExpiration(String.valueOf(3000));
        Message message = defaultRabbitTemplate
                .getMessageConverter()
                .toMessage(data, messageProperties);

       defaultRabbitTemplate.convertAndSend(RabbitConfig.NORMAL_EXCHANGE_A,
                RabbitConfig.NORMAL_ROUTING_KEY_A,
                message);
    }
}

2.4. 代码（消费者）

消费者的逻辑比较简单，次要是别离监听业务队列和死信队列，这里将两个队列的音讯输入日志。这里模仿的是消费者 nack 音讯，并且不退回队列的状况。

MessageListener.java

/**
 * @description:
 * @date: 2020/7/12 11:07 下午
 * @author: kerry
 */
@Component
@Slf4j
public class MessageListener {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private ObjectMapper objectMapper;

    /**
     * @param message
     * @param channel
     * 监听 业务队列
     *
     * @throws Exception
     */
    @RabbitListener(queues = RabbitConfig.NORMAL_QUEUE_A)
    @RabbitHandler
    public void onMessage(Message message, Channel channel) throws Exception {String contentType = message.getMessageProperties().getContentType();
        String bodyText = null;
        switch (contentType) {
            // 字符串
            case MessageProperties.CONTENT_TYPE_TEXT_PLAIN:
                bodyText = (String) rabbitTemplate.getMessageConverter().fromMessage(message);
                break;
            //json 对象
            case MessageProperties.CONTENT_TYPE_JSON:
                User user = objectMapper.readValue(message.getBody(), User.class);
                bodyText = user.toString();
                break;
        }
        log.info("业务队列 - 回绝音讯:" + bodyText);
        channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
         /**
         * 提早队列，被生产的音讯须要重回队列
         */
        //channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);
    }


    /**
     * @param message
     * @param channel
     * 监听 死信队列
     *
     * @throws Exception
     */
    @RabbitListener(queues = RabbitConfig.DEAD_LETTER_QUEUE_A)
    @RabbitHandler
    public void onMessageDLX(Message message, Channel channel) throws Exception {String contentType = message.getMessageProperties().getContentType();
        String bodyText = null;
        switch (contentType) {
            // 字符串
            case MessageProperties.CONTENT_TYPE_TEXT_PLAIN:
                bodyText = (String) rabbitTemplate.getMessageConverter().fromMessage(message);
                break;
            //json 对象
            case MessageProperties.CONTENT_TYPE_JSON:
                User user = objectMapper.readValue(message.getBody(), User.class);
                bodyText = user.toString();
                break;
        }
        log.info("死信队列 - 接管音讯:" + bodyText);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
 
    }

}

2.5. 剖析

结尾说过，导致音讯转为死信队列的形式有 3 种，上面就从代码中剖析这 3 种状况。

咱们回过头来看看消费者这边，定义业务队列的办法：

    /**
     * NORMAL 队列
     * @return
     */
    @Bean
    public Queue QueueA(){
        return QueueBuilder
                .durable(NORMAL_QUEUE_A)
                // 设置 死信交换机
                .withArgument(X_DEAD_LETTER_EXCHANGE,DEAD_LETTER_EXCHANGE_A)
                .withArgument(X_DEAD_LETTER_ROUTING_KEY,DEAD_LETTER_ROUTING_KEY_A)
                // 设置 队列所有音讯 存活工夫 8 秒
                .withArgument(X_MESSAGE_TTL,8000)
                // 设置 队列最大长度 10 条
                .withArgument(X_MAX_LENGTH,10)
                .build();}

1. nack 回绝音讯，requeue=false

在死信队列的架构中，只有在业务队列中设置了死信交换机 x-dead-letter-exchange。消费者代码中，咱们在业务队列的生产过程中 nack 音讯，并且 requeue=false 即可。x-dead-letter-routing-key 能够不设置，如果不设置，默认取音讯原来的路由键。
代码如下：

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

channel.basicNack 办法的参数如下:

  1. deliveryTag: 该音讯的 index。
  2. multiple：是否批量.true: 将一次性回绝所有小于 deliveryTag 的音讯。
  3. requeue：被回绝的是否从新入队列。

2. 音讯超时 TTL

TTL（Time-To-Live）指音讯的存活工夫，咱们有两种形式设置音讯的 TTL：

  1. 设置队列的 TTL，即参数x-message-ttl，设置后，进入该队列的所有音讯的 TTL 都为对应的值。
  2. 设置单个音讯的 TTL，在生产者代码中，给音讯的属性中设置过期工夫。如：messageProperties.setExpiration(String.valueOf(3000));，就是设置音讯的 TTL 为 3 秒。

不过要留神的是，如果曾经设置了 nack 的死信逻辑，TTL 的死信就不失效了。情理也很简略，因为 nack 音讯和 requeue=false 一起用，代表音讯被生产了，并且音讯不会重回队列，间接被抛弃或进入死信队列，又怎么会在队列中超时了呢。

3. 超过队列长度

能够设置队列长度，例如最大接管音讯的数量。当音讯在队列中曾经达到最大数量，那么前面再来的音讯，就会被间接丢进死信队列。
咱们也是中定义业务队列的代码中，有通过 x-max-length 参数，设置业务队列的长度。

3. 提早队列

在我还不晓得提早队列之前，我就感觉消息中间件应该具备这样的性能。在音讯公布到队列后，我冀望每个音讯提早指定工夫后，再被消费者获取到。例如：在领取模块中，当用户生成订单，再到领取实现订单，是有一段时间的。而咱们个别会给这个订单设置超时工夫，如果超过了这段时间，订单就应该被勾销，无奈再领取。那么将订单作为音讯，就能够利用提早队列来实现勾销订单的逻辑。

RabbitMQ 并不间接反对提早队列的性能，而是作为一个概念，你能够利用死信队列来实现“提早队列”。利用 TTL 超时工夫的死信形式，来实现提早队列。

回顾一下上段中 TTL 的形式，咱们在业务队列中除了定义死信交换机x-dead-letter-exchange，还能够定义队列的生存工夫x-message-ttl，或者设置音讯的过期工夫。而如果咱们不生产这个业务队列中的音讯，那么音讯在达到 TTL 后，就会主动转到死信队列中。如果咱们只生产死信队列中的音讯，疏忽掉业务队列这个“中转站”，就相当于音讯在被公布后，通过指定时间延迟，在死信队列中被生产，这就造成了一个“提早队列”。

因为提早队列就是死信队列的一种实现，所以代码层面上能够间接参考上段中 TTL 的局部。

4. 优先队列

优先队列，顾名思义，领有高优先级的队列具备高的优先权，优先级高的音讯具备优先被生产的势力。因而在优先队列中有两个逻辑点：队列是优先队列 ， 音讯有优先级。可参考死信队列章节中，TTL 局部的代码，上面对代码改变中央做一下阐明：

  1. 设置队列最大优先级，即音讯可应用的最大优先级数字，可通过 x-max-priority 参数来设置。
  2. 设置音讯的优先级，在生产者代码中，在音讯的属性中设置优先级，优先级越大，越先被生产。如：messageProperties.setPriority(3)，即设置该音讯的优先级为 3。

优先队列的应用场景是在：音讯有分优先级的需要，并且并发量较大。要求并发量大，是因为如果所有音讯在公布之后，马上就被生产了，那么分优先级的必要性就不大了。

退出移动版