RabbitMQ高级特性

MQ可能存在的问题

消息是否真的发送出去了

消息发送后，发送端不知道RabbitMQ是否真的收到了消息

若RabbitMQ异常，消息丢失后，订单处理流程停止，业务异常

需要使用RabbitMQ发送端确认机制，确认消息发送

消息真的被路由了吗

消息发送后，发送端不知道消息是否被正确路由，若路由异常，消息会被丢弃

消息丢弃后，订单处理流程停止，业务异常

需要使用RabbitMQ消息返回机制，确认消息被正确路由

消费端处理异常怎么办

默认情况下，消费端接收消息时，消息会被自动确认（ACK）

消费端消息处理异常时，发送端与消息中间件无法得知消息处理情况

需要使用RabbitMQ消费端确认机制，确认消息被正确处理

消费端处理的过来吗

业务高峰期，可能出现发送端与接收端性能不一致，大量消息被同时推送给接收端，造成接收端服务崩溃

需要使用RabbitMQ消费端限流机制，限制消息推送速度，保障接收端服务稳定

队列爆满怎么办

默认情况下，消息进入队列，会永远存在，直到被消费

大量堆积的消息会给RabbitMQ产生很大的压力

需要使用RabbitMQ消息过期时间机制，防止消息大量积压

如何转移过期消息

消息被设置了过期时间，过期后会被直接丢弃

直接被丢弃的消息，无法对系统运行异常发出警报

需要使用RabbitMQ死信队列，收集过期消息，以供分析

发送方

需要使用RabbitMQ发送端确认机制，确认消息成功发送到RabbitMQ并被处理
需要使用RabbitMQ消息返回机制，若没有发现目标队列，中间件会通知发送方

发送端确认（消息真的发出去了吗）

消息发送后，若中间件收到消息，会给发送端一个应答

三种确认机制

单条同步确认

配置channel，开启确认模式：channel.confirmSelect()

每发送一条消息，调用channel.waitForConfirms()方法，等待确认

多条同步确认（累积确认）

配置channel，开启确认模式：channel.confirmSelect()

发送多条消息后，调用channel.waitForConfirms()方法，等待确认

异步确认

配置channel，开启确认模式：channel.confirmSelect()

在channel上添加监听：addConfirmListener，发送消息后，会回调此方法，通知是否发送成功

异步确认有可能是单条，也有可能是多条，取决于MQ
channel.confirmSelect();

for (int i = 0; i < 50; i++) {
    channel.basicPublish("exchange.order.restaurant", "key.restaurant", null, messageToSend.getBytes());
    log.info("message send");
}

// 多条同步确认
if (channel.waitForConfirms()) {
    log.info("ACK Success~");
} else {
    log.info("ACK Error~");
}
channel.confirmSelect();

// 异步确认
channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleAck(long deliveryTag, boolean multiple) throws IOException {
        log.info("ACK, deliveryTag: {}, multiple: {}", deliveryTag, multiple);
    }

    @Override
    public void handleNack(long deliveryTag, boolean multiple) throws IOException {
        log.info("NACK, deliveryTag: {}, multiple: {}", deliveryTag, multiple);
    }
});

消息返回机制（消息真的被路由了吗）

消息发送后，中间件会对消息进行路由
若没有发现目标队列，中间件会通知发送方
Return Listener会被调用

具体实现

在RabbitMQ基础配置中有一个关键配置项：Mandatory
Mandatory若为false，RabbitMQ将直接丢弃无法路由的消息
Mandatory若为true，RabbitMQ才会处理无法路由的消息

// 路由异常时，消息返回机制
// ReturnCallback是对ReturnListener参数的封装
channel.addReturnListener(new ReturnCallback() {
    @Override
    public void handle(Return returnMessage) {
        log.info("Message Return -> returnMessage:{}", returnMessage.toString());
        // 相应的业务操作
        // ......
    }
});
// 路由异常时，消息返回机制
channel.addReturnListener(new ReturnListener() {
    @Override
    public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
        log.info("Message Return -> " +
                        "replyCode:{}, replyText:{}, exchange:{}, routingKey:{}, properties:{}, body:{}",
                replyCode, replyText, exchange, routingKey, properties, new String(body));
        // 相应的业务操作
        // ......
    }
});

// 参数表：String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body
channel.basicPublish("exchange.order.restaurant", "key.order", true, null, messageToSend.getBytes());

消费方

需要使用RabbitMQ消费端确认机制，确认消息没有发生处理异常
需要使用RabbitMQ消费端限流机制，限制消息推送速度，保障接收端服务稳定

消费端ACK

自动ACK：消费端收到消息后，会自动签收消息

手动ACK：消费端收到消息后，不会自动签收消息，需要在业务代码中显式签收消息

若RabbitMQ将消息发送给消费方后，在等待ACK时，消费方下线，待ACK的消息又会重新处于Ready状态

手动ACK类型

单条手动ACK：multiple = false

多条手动ACK：multiple = true

推荐使用单条ACK

手动NACK后重回队列

若设置了重回队列，消息被NACK后，会返回队列末尾，等待进一步被处理

一般不建议开启重回队列，因为第一次处理异常的消息，再次处理，基本上也是异常
// 多条ACK
if (message.getEnvelope().getDeliveryTag() % 5 == 0) {
    channel.basicAck(message.getEnvelope().getDeliveryTag(), true);
}
// 单条ACK
channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
// NACK，重新入队
channel.basicNack(message.getEnvelope().getDeliveryTag(), false, true);

消费端限流（消费端处理）

实际场景：

业务高峰期，某个微服务崩溃了，期间队列积压了大量消息，服务重新上线后，收到大量并发消息

将同样多的消息推送给能力不同的副本，会导致部分副本异常

RabbitMQ - QoS

针对上述问题，RabbitMQ开发了QoS（服务质量保证）功能

QoS功能保证了在一定数目的消息未被确认前，不消费新的消息

QoS的前提是不使用自动确认

QoS原理

QoS原理是消费端有一定数量的消息未被ACK确认时，RabbitMQ不给消费端推送新的消息

RabbitMQ使用QoS机制来实现消费端限流

消费端限流机制参数设置

perfetchCount：针对一个消费端最多推送多少未确认消息

global：true，针对整个消费端限流；false，针对当前channel限流

prefetchSize：0（单个消息大小限制，一般为0）

prefetchSize与global两项（AMQP协议中的内容），RabbitMQ暂时未实现

QoS最大的作用是便于服务的临时水平扩展，RabbitMQ会将待消费的消息“一点点”的发送给消费方，临时扩展的服务也能分得消息。若未使用QoS，RabbitMQ会将待消费的消息一次全部发给现有的服务，然后等待消息ACK；开启QoS，待消费的消息为Ready状态，可以分给新上线的服务。
channel.basicQos(2);
// 参数表：String queue, boolean autoAck, DeliverCallback deliverCallback，...
channel.basicConsume("queue.restaurant", false, deliverCallback, consumerTag -> {});

RabbitMQ自身

大量的消息堆积会给RabbitMQ产生很大的压力，需要使用RabbitMQ消息过期时间，防止消息大量积压
消息过期后会被直接丢弃，无法对系统异常发出警报，需要使用RabbitMQ死信队列，收集过期消息，以供分析

消息的过期时间（TTL）

RabbitMQ的过期时间称为TTL（Time to Live），生存时间

RabbitMQ的过期时间分为消息TTL和队列TTL

消息TTL设置了单条消息的过期时间

队列TTL设置了队列中所有消息的过期时间

设置合适的TTL

TTL的设置主要考虑技术架构与业务

TTL应该明显长于服务的平均重启时间

建议TTL长于业务高峰期时间
// 消息TTL
// 构造者模式
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
    	// 单位：ms
        .expiration("15000")
        .build();
channel.basicPublish("exchange.order.restaurant", "key.restaurant", 
        properties, messageToSend.getBytes());
// 队列TTL
Map<String, Object> args = new HashMap<>();
// 队列中的消息的过期时间（ms）
args.put("x-message-ttl", 15000);
// 队列自身过期时间
// args.put("x-expires", 10000);
// 参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> args
channel.queueDeclare("queue.restaurant", true, false, false, args);

死信队列

队列被配置了DLX属性（Dead-Letter-Exchange）

当一个消息变成死信（dead message）后，能重新被发布到另一个Exchange，这个Exchange也是一个普通交换机

死信被死信交换机路由后，一般进入一个固定队列

怎样变成死信

消息被拒绝（reject/nack）并且requeue=false

消息过期（TTL到期）

队列达到最大长度

死信队列设置

设置转发、接收死信的交换机和队列

Exchange: dlx.exchange

Queue: dlx.queue

RoutingKey: #

在需要设置死信的队列加入参数：x-dead-letter-exchange = dlx.exchange
// 声明转发死信的交换机，队列，建立绑定关系
channel.exchangeDeclare(
    "exchange.dlx",
    BuiltinExchangeType.TOPIC,
    true, false, null);
channel.queueDeclare(
    "queue.dlx",
    true, false, false, null);
channel.queueBind("queue.dlx", "exchange.dlx", "#");
// 声明死信队列
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "exchange.dlx");
args.put("x-message-ttl", 15000);
args.put("x-max-length", 3);
channel.queueDeclare("queue.restaurant", true, false, false, args);

善用RabbitMQ高级特性

对于RabbitMQ的高级特性，要善加利用
接收端确认、死信队列是非常常用的特性

慎用RabbitMQ高级特性

不要无限追求高级，用上所有RabbitMQ的高级特性
重回队列、发送端确认是不常用的特性，谨慎使用

善用RabbitMQ管控台

管控台是调试的利器
RabbitMQ高级特性多数都涉及交换机、队列的属性配置，可以在管控台确认配置是否生效
RabbitMQ高级特性很多都可以在管控台进行试验