RabbitMQ高级

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

• confirm 确认模式

• return 退回模式

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

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

• 消息从 producer 到 exchange 则会返回一个 confirmCallback 。

• 消息从 exchange 到 queue 投递失败则会返回一个 returnCallback 。

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

confirm确认模式代码实现(这个是保证消息生产者顺利的将消息发送到交换机)
return回退模式： 当消息发送给Exchange后，Exchange路由到Queue失败是 才会执行 ReturnCallBack

小结

对于确认模式：

• 设置ConnectionFactory的publisher-confirms=“true” 开启 确认模式。

• 使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。

对于退回模式

• 设置配置文件中的ConnectionFactory的publisher-returns=“true” 开启 退回模式。

• 使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后，如果设置了rabbitTemplate.setMandatory(true)参数，则会将消息退回给producer。并执行回调函数returnedMessage。

Consumer ACK

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

有三种确认方式：

• 自动确认：acknowledge=“none”

• 手动确认：acknowledge=“manual”

• 根据异常情况确认：acknowledge=“auto”，（这种方式使用麻烦，不作讲解）

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

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

消费端限流

• 在rabbit:listener-container 中配置 prefetch属性设置消费端一次拉取多少消息

• 消费端的确认模式一定为手动确认。acknowledge=“manual”

TTL

TTL 全称 Time To Live（存活时间/过期时间）。当消息到达存活时间后，还没有被消费，会被自动清除。

RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。
设置队列的过期时间
在消息的生产方中，在 spring-rabbitmq-producer.xml 配置文件中，添加如下配置：

<!--ttl-->
<rabbit:queue name="test_queue_ttl" id="test_queue_ttl">
    <!--设置queue的参数-->
    <rabbit:queue-arguments>
        <!--x-message-ttl指队列的过期时间-->
        <entry key="x-message-ttl" value="100000" value-type="java.lang.Integer"/>
    </rabbit:queue-arguments>
</rabbit:queue>

<rabbit:topic-exchange name="test_exchange_ttl" >
    <rabbit:bindings>
        <rabbit:binding pattern="ttl.#" queue="test_queue_ttl"></rabbit:binding>
    </rabbit:bindings>
</rabbit:topic-exchange>
1
2
3
4
5
6
7
8
9
10
11
12
13
14

设置单个消息的过期时间

@Test
public void testTtl() {

  // 消息后处理对象，设置一些消息的参数信息
    MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {

        @Override
        public Message postProcessMessage(Message message) throws AmqpException {
            //1.设置message的信息
            message.getMessageProperties().setExpiration("5000");//消息的过期时间
            //2.返回该消息
            return message;
        }
    };

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

    for (int i = 0; i < 10; i++) {
        if(i == 5){
            //消息单独过期
            rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....",messagePostProcessor);
        }else{
            //不过期的消息
            rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....");

        }

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

如果设置了消息的过期时间，也设置了队列的过期时间，它以时间短的为准。

• 队列过期后，会将队列所有消息全部移除。
• 消息过期后，只有消息在队列顶端，才会判断其是否过期(移除掉)

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

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

1. 队列消息长度到达限制；

2. 消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；

3. 原队列存在消息过期设置，消息到达超时时间未被消费；

队列绑定死信交换机：

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