[大数据] Rabbit高级特性如何理解？

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> Rabbit高级特性如何理解？ -> 正文阅读

[大数据]Rabbit高级特性如何理解？

Rabbit高级特性

1.生产者确认

confirm确认模式、return回退模式

目标：演示消息确认模式效果
- 生产者发布消息确认模式特点，不论消息是否进入交换机均执行回调方法
实现步骤：
1. 在配置文件中，开启生产者发布消息确认模式
2. 编写生产者确认回调方法
3. 在RabbitTemplate中，设置消息发布确认回调方法
4. 请求测试
  - 测试成功回调
  - 测试失败回调

# 开启生产者确认模式：(confirm),投递到交换机，不论失败或者成功都回调
spring.rabbitmq.publisher-confirms=true
# 开启生产者回退模式:(returns)，交换机将消息路由到队列，出现异常则回调
spring.rabbitmq.publisher-returns=true

@Component
public class MessageConfirmCallback implements RabbitTemplate.ConfirmCallback , RabbitTemplate.ReturnCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 创建RabbitTemplate对象之后执行当前方法，为模板对象设置回调确认方法
     * 设置消息确认回调方法
     * 设置消息回退回调方法
     */
    @PostConstruct
    public void initRabbitTemplate(){
        rabbitTemplate.setConfirmCallback(this::confirm);
        rabbitTemplate.setReturnCallback(this::returnedMessage);
    }


    /**
     * @param correlationData  投递相关数据
     * @param ack  是否投递到交换机
     * @param cause  投递失败原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            System.out.println("投递成功");
        } else {
            System.out.println("投递失败");
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("交换机至消息队列出错：");
        System.out.println("交换机："+exchange);
        System.out.println("路由键："+routingKey);
        System.out.println("错误状态码："+replyCode);
        System.out.println("错误原因："+replyText);
        System.out.println("发送消息内容："+message.toString());
    }
}

2. 消费者确认（ACK）

是消费端收到消息的一种确认机制
三种类型
1. 自动确认：acknowledge="none"
2. 手动确认：acknowledge="manual"
3. 根据异常情况确认：acknowledge="auto"
其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息丢失

配置手动确认

# 配置开启手动签收
# 简单模式的开启手动签收
spring.rabbitmq.listener.simple.acknowledge-mode=manual
# 路由模式开启手动签收
spring.rabbitmq.listener.direct.acknowledge-mode=manual
# 是否支持重试
spring.rabbitmq.listener.direct.retry.enabled=true

@Component
@RabbitListener(queues = "routing_queue1")
public class AckConsumerListerner {


    @RabbitHandler
    public void ququeListernerHandler(String msg, Message message, Channel channel) throws IOException {
        System.out.println("收到的消息");
        MessageProperties messageProperties = message.getMessageProperties();
        long deliveryTag = messageProperties.getDeliveryTag();
        try {
            if (msg.contains("哈哈")) {
                throw new RuntimeException("不允许哈哈笑");
            }
            /**
             * 手动签收消息
             * 参数1：消息投递标签
             * 参数2：是否批量签收：true一次性签收，效率高。false,只签收当前消息
             */
            channel.basicAck(deliveryTag,false);
        } catch (IOException e) {
            e.printStackTrace();
            /**
             * 手动拒绝签收
             * 参数1：当前消息的投递标签
             * 参数2：是否批量签收：true一次性签收所有，false，只签收当前消息
             * 参数3：是否重回队列，true为重回队列，false为不重回
             */
            channel.basicNack(deliveryTag,false,true);
        }

    }

}

消费端限流

rabbit消费端限流.png

场景

A系统中需要维护相关的业务功能，导致大量的消息都会在MQ中累积，A系统成功启动后，消费者会一次性将MQ中累积的大量的消息拉到自己的服务，导致服务在短时间内会处理大量的业务，可能会导致系统服务的崩溃
当大量用户请求过来的时候，消费端限流也非常必要；

代码实现

# 设置消费端限流，每次拉取消息多少条，默认是250条
spring.rabbitmq.listener.direct.prefetch=1
复制代码

TTL(消息存活时间)Time To Live

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

rabbit消息存活时间.png

实现
- 设置队列的时候设置Arguments，添加x-message-ttl = 5000 (毫秒)

死信队列

rabbit死信队列.png

当消息成为Dead message后，可以被重新发送到另一个交换机，这个交换机就是Dead Letter Exchange（死信交换机简写：DLX）。
消息成为死信的三种情况
- 队列消息长度到达限制；
- 消费者拒接消息(basicNack)，并且不把消息重新放回源队列，requeue=false；
- 源队列存在消息过期设置，消息到达超时时间未被消费；
实现死信队列
- 创建死信队列deadQueue Type : Classic
- 创建死信交换机deadExchange：Type : direct
- 死信交换机绑定死信队列 Rounting key: order.dead
- 给目标队列设置Arguments参数：
  - x-dead-letter-exchange = deadExchange
  - x-dead-letter-routing-key = order.dead