RabbitMQ-高级

立聪堂德州十三局店  金牌会员 | 2024-8-21 12:09:21 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 511|帖子 511|积分 1533

RabbitMQ-高级


  
1 发送者的可靠性

1.1 发送者重连

有的时间由于网络颠簸，大概会出现发送者连接MQ失败的情况，通过配置我们可以开启连接失败后的重连机制：

注意： 当网络不稳定的时间，利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试，也就是说多次重试等候的过程中，当前线程是被阻塞的，会影响业务性能。如果对于业务性能有要求，发起禁用重试机制。如果一定要利用，请合理配置等候时长和重试次数，当然也可以考虑利用异步线程来实行发送消息的代码。
1.2 发送者确认

SpringAMOP提供了 Publisher Confirm 和 Publisher Return 两种确认机制。开启确机制认后，当发送者发送消息给MQ后，MO会返回确认效果给发送者。返回的效果有以下几种情况：


  • 消息投递到了MQ，但是路由失败。此时会通过PublisherReturn返回路由非常缘故原由，然后返回ACK，告知投递成功（这种错误是程序员造成了，是路由代码的错误，不是MQ的题目）
  • 临时消息（非持久化，没有写入磁盘）投递到了MQ，而且入队成功，返回ACK，告知投递成功。
  • 持久消息投递到了MQ，而且入队完成持久化，返回ACK，告知投递成功。
  • 其它情况都会返回NACK，告知投递失败

1.2.1 SpringAMQP实现发送者确认


  • 在publisher这个微服务的application.yml中添加配置

​ 配置说明：


  • 这里publisher-confirm-type有三种模式可选：

    • none：关闭confirm机制
    • simple：同步阻塞等候MQ的回执消息
    • correlated：MQ异步回调方式返回回执消息


  • 每个RabbitTemplate只能配置一个ReturnCallback，因此需要在项目启动过程中配置：


  • 发送消息，指定消息ID、消息ConfirmCallback

    发送成功与否都会实行 ConfirmCallback 发送失败才会实行 ReturnCallback 而且ReturnCallback只需要初始化一次就行，以是就在配置类中初始化一次，利用PostConstruct
    发送者：
    1. @Slf4j
    2. @Configuration
    3. @RequiredArgsConstructor
    4. public class MqConfig {
    6.     private final RabbitTemplate rabbitTemplate;
    8.     @PostConstruct
    9.     public void init(){
    10.         rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
    11.             @Override
    12.             public void returnedMessage(ReturnedMessage returnedMessage) {
    13.                 log.error("监听到了消息return callback");
    14.                 log.debug("exchange：{}",returnedMessage.getExchange());
    15.                 log.debug("routingKey：{}",returnedMessage.getRoutingKey());
    16.                 log.debug("message：{}",returnedMessage.getMessage());
    17.                 log.debug("replyCode：{}",returnedMessage.getReplyCode());
    18.                 log.debug("replyText：{}",returnedMessage.getReplyText());
    19.             }
    20.         });
    21.     }
    22. }
    复制代码
    1. @Test
    2.     public void testConfirmCallback()  throws InterruptedException {
    3.         //0. 创建correlationData
    4.         CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
    5.         cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
    6.             @Override
    7.             public void onFailure(Throwable ex) {
    8.                 log.error("Spring amqp 处理确认结果异常",ex);
    9.             }
    11.             @Override
    12.             public void onSuccess(CorrelationData.Confirm result) {
    13.                 // 判断是否成功
    14.                 if(result.isAck()){
    15.                     //成功
    16.                     log.debug("收到ConfirmCallback ack，消息发送成功");
    17.                 }else {
    18.                     //失败
    19.                     log.error("收到ConfirmCallback nack，消息发送失败！reason:{}",result.getReason());
    20.                 }
    21.             }
    22.         });
    24.         //1. 交换机名
    25.         String exchangeName = "hmall.direct";
    26.         //2. 消息
    27.         String message = "hello,Spring amqp";
    28.         //3. 发送消息
    29.         rabbitTemplate.convertAndSend(exchangeName, "red", message + "red",cd);
    30.         rabbitTemplate.convertAndSend(exchangeName, "yellow", message + "yellow",cd);
    31.         rabbitTemplate.convertAndSend(exchangeName, "blue2", message + "blue",cd);        //路由失败
    32.         Thread.sleep(2000);
    33.     }
    复制代码
    效果：

2 MQ的可靠性

在默认情况下，RabbitMQ会将吸收到的信息保存在内存中以降低消息收发的耽误。这样会导致两个题目：


  • 一旦MQ宕机，内存中的消息将会丢失
  • 内存空间有限，当斲丧者故障或处置惩罚过慢时，会导致消息积压，引发MQ阻塞。
2.1 数据持久化

RabbitMQ实现数据持久化包括三个方面：


  • 交换机持久化
  • 队列持久化
  • 消息持久化（如果交换机队列是持久的，但是消息是临时，那么MQ重启后消息也会丢失）

java代码：将消息转化为非持久的
  1.     @Test
  2.     public void testSendMessage() {
  3.         //1. 自定义构建消息
  4.         Message msg = MessageBuilder.withBody("hello,SpringAMQP".getBytes(StandardCharsets.UTF_8))
  5.                 .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)        //非持久
  6.                     //.setDeliveryMode(MessageDeliveryMode.PERSISTENT)        //持久
  7.                 .build();
  8.         //2. 发送消息
  9.         rabbitTemplate.convertAndSend("object.queue",msg);
  10.     }
复制代码
2.2 Lazy Queue

在默认情况下，RabbitMQ会将吸收到的信息保存在内存中以降低消息收发的耽误。但在某些特殊情况下，这会导致消息积压，比如：


  • 斲丧者宕机或出现网络故障
  • 消息发送量激增，凌驾了斲丧者处置惩罚速度
  • 斲丧者处置惩罚业务发生阻塞
一旦出现消息堆积题目，RabbitMQ的内存占用就会越来越高，直到触发内存预警上限。此时RabbitMQ会将内存消息刷到磁盘上，这个行为成为PageOut. PageOut会泯灭一段时间，而且会阻塞队列进程。因此在这个过程中RabbitMQ不会再处置惩罚新的消息，生产者的全部请求都会被阻塞。
为了解决这个题目，从RabbitMQ的3.6.0版本开始，就增长了Lazy Queues的模式，也就是惰性队列。惰性队列的特性如下：


  • 吸收到消息后直接存入磁盘而非内存
  • 斲丧者要斲丧消息时才会 从磁盘中读取并加载到内存（也就是写得快，但是懒加载，不过可以提前缓存部分消息到内存，最多2048条）
  • 支持数百万条的消息存储
而在3.12版本之后，LazyQueue已经成为全部队列的默认格式。因此官方推荐升级MQ为3.12版本或者全部队列都设置为LazyQueue模式。
控制台利用： 要设置一个队列为惰性队列，只需要在声明队列时，指定x-queue-mode属性为lazy即可：

Java客户端利用：



3 斲丧者的可靠性

3.1 斲丧者确认机制

斲丧者确实机制（Consumer Acknowledgement）是为了确认斲丧者是否成功处置惩罚消息。当斲丧者处置惩罚消息竣事后，应该向RabbitMQ发送一个回执，告知RabbitMQ本身消息处置惩罚状态：


  • ack：成功处置惩罚消息，RabbitMQ从队列中删除该消息
  • nack：消息处置惩罚失败，RabbitMO需要再次投递消息
  • reject：消息处置惩罚失败并拒绝该消息，RabbitMQ从队列中删除该消息

SpringAMQP已经实现了消息确认功能。并允许我们通过配置文件选择ACK处置惩罚方式，有三种方式:


  • none:不处置惩罚。即消息投递给斲丧者后立即ack，消息会立即从MQ删除。非常不安全，不发起利用
  • manual:手动模式。需要本身在业务代码中调用api，发送ack或reject，存在业务入侵，但更灵活
  • auto:自动模式。SpringAMQP利用AOP对我们的消息处置惩罚逻辑做了围绕增强，当业务正常实行时则自动返回ack.
    当业务出现非常时，根据非常判定返回不同效果:

    • 如果是业务非常，比如RuntimeException，会自动返回nack
    • 如果是消息处置惩罚或校验非常，比如MessageConversionException，自动返回reject


3.2 失败重试策略

SpringAMQP提供了斲丧者失败重试机制，在斲丧者出现非常时利用本地重试，而不是无限的requeue到mg。我们可以通过在application.yaml文件中添加配置来开启重试机制：

在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有MessageRecoverer接口来处置惩罚，它包含三种不同的实现:


  • RejectAndDontRequeueRecoverer:重试耗尽后，直接reject，丢弃消息。默认就是这种方式
  • ImmediateRequeueMessageRecoverer:重试耗尽后，返回nack，消息重新入队
  • RepublishMessageRecoverer:重试耗尽后，将失败消息投递到指定的交换机

失败消息处置惩罚策略
将失败处置惩罚策略改为RepublishMessageRecoverer：


  • 首先，定义吸收失败消息的交换机、队列及其绑定关系，此处略：
  • 然后，定义RepublishMessageRecoverer：

  1. @Configuration
  2. @RequiredArgsConstructor
  3. public class ErrorMessageConfiguration {
  5.    private final RabbitTemplate rabbitTemplate;
  7.    @Bean
  8.    public DirectExchange errorExchange(){
  9.        return new DirectExchange("error.direct");
  10.    }
  12.    @Bean
  13.    public Queue errorQueue(){
  14.        return new Queue("error.queue");
  15.    }
  17.    @Bean
  18.    public Binding errorQueueBinding(){
  19.        return BindingBuilder.bind(errorQueue()).to(errorExchange()).with("error");
  20.    }
  22.    @Bean
  23.    public MessageRecoverer messageRecoverer(){
  24.        return new RepublishMessageRecoverer(rabbitTemplate,"error.direct","error");
  25.    }
  26. }
复制代码

3.3 业务幂等性

幂等是一个数学概念，用函数表达式来形貌是这样的：f(x) = f(f(x))，例如求绝对值函数。
在程序开发中，则是指同一个业务，实行一次或多次对业务状态的影响是同等的。例如：


  • 根据id删除数据
  • 查询数据
  • 新增数据
但数据的更新每每不是幂等的，如果重复实行大概造成不一样的后果。比如：


  • 取消订单，恢复库存的业务。如果多次恢复就会出现库存重复增长的情况
  • 退款业务。重复退款对商家而言会有经济丧失。
以是，我们要尽大概避免业务被重复实行。
然而在实际业务场景中，由于意外常常会出现业务被重复实行的情况，例如：


  • 页面卡顿时频繁刷新导致表单重复提交
  • 服务间调用的重试
  • MQ消息的重复投递
以是，我们要尽大概避免业务被重复实行。
然而在实际业务场景中，由于意外常常会出现业务被重复实行的情况，例如：


  • 页面卡顿时频繁刷新导致表单重复提交
  • 服务间调用的重试
  • MQ消息的重复投递
我们在用户支付成功后会发送MQ消息到生意业务服务，修改订单状态为已支付，就大概出现消息重复投递的情况。如果斲丧者不做判定，很有大概导致消息被斲丧多次，出现业务故障。
举例：

  • 假如用户刚刚支付完成，而且投递消息到生意业务服务，生意业务服务更改订单为已支付状态。
  • 由于某种缘故原由，例如网络故障导致生产者没有得到确认，隔了一段时间后重新投递给生意业务服务。
  • 但是，在新投递的消息被斲丧之前，用户选择了退款，将订单状态改为了已退款状态。
  • 退款完成后，新投递的消息才被斲丧，那么订单状态会被再次改为已支付。业务非常。
因此，我们必须想办法保证消息处置惩罚的幂等性。这里给出两种方案：


  • 唯一消息ID
  • 业务状态判定
3.3.1 唯一消息id

方案一，是给每个消息都设置一个唯一id，利用id区分是否是重复消息


  • 每一条消息都生成一个唯一的id，与消息一起投递给斲丧者。
  • 斲丧者吸收到消息后处置惩罚本身的业务，业务处置惩罚成功后将消息ID保存到数据库
  • 如果下次又收到相同消息，去数据库查询判定是否存在，存在则为重复消息放弃处置惩罚。
SpringAMQP的MessageConverter自带了MessageID的功能，我们只要开启这个功能即可。
以Jackson的消息转换器为例：
  1. @Bean
  2. public MessageConverter messageConverter(){
  3.     // 1.定义消息转换器
  4.     Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
  5.     // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
  6.     jjmc.setCreateMessageIds(true);
  7.     return jjmc;
  8. }
复制代码
此时吸收对象可以直接用Message吸收

3.3.2 业务判定

方案二：是结合业务逻辑，基于业务本身做判定。以余额支付业务为例：


java代码多了第1、2步
  1.     public void listenPaySuccess(Long orderId){
  2.         //1.查询订单
  3.         Order order = orderService.getById(orderId);
  4.         //2. 判断订单状态 是否为未支付
  5.         if(order == null || order.getStatus() != 1){
  6.             //不做处理
  7.             return;
  8.         }
  9.         //3. 标记订单状态为已支付
  10.         orderService.markOrderPaySuccess(orderId);
  11.     }
复制代码

4 耽误消息

耽误消息： 发送者发送消息时指定一个时间，斲丧者不会立即收到消息，而是在指定时间之后才收到消息。
耽误任务： 设置在一定时间之后才实行的任务。

4.1 死信交换机

当一个队列中的消息满足下列情况之一时，就会成为死信(dead letter)：


  • 斲丧者利用basic.reject或 basic.nack声明斲丧失败，而且消息的requeue参数设置为false
  • 消息是一个逾期消息(达到了队列或消息本身设置的逾期时间)，超时无人斲丧
  • 要投递的队列消息堆积满了，最早的消息大概成为死信
如果 队列 通过dead-letter-exchange属性指定了一个 交换机 ，那么该队列中的死信就会投递到这个交换机中。这个交换机 称为死信交换机(Dead LetterExchange，简称DLX)。

dlx交换机代码
  1.     @RabbitListener(bindings = @QueueBinding(
  2.             value = @Queue(name = "dlx.queue", durable = "true"),
  3.             exchange = @Exchange(name = "dlx.direct", type = ExchangeTypes.DIRECT),
  4.             key = {"hi"}
  5.     ))
  6.     public void listendlxQueue(String message) {
  7.         log.info("消费者监听到dlx.queue的消息：{}", message);
  9.     }
复制代码
normal交换机
  1. @Slf4j
  2. @Configuration
  3. public class NormalConfiguration {
  5.     @Bean
  6.     public DirectExchange normalExchange(){
  7.         return new DirectExchange("normal.direct");
  8.     }
  10.     @Bean
  11.     public Queue normalQueue(){
  12.         return QueueBuilder.durable("normal.queue").deadLetterExchange("dlx.direct").build();
  13.     }
  15.     @Bean
  16.     public Binding normalExchangeBinding(Queue normalQueue,DirectExchange normalExchange){
  18.         return BindingBuilder.bind(normalQueue).to(normalExchange).with("hi");
  19.     }
  21. }
复制代码
值得一提的是，两个交换机中绑定的key要同等，比如例中都是"hi"
发送方添加了耽误时间的消息
  1.     @Test
  2.     public void testSendDelayMessage() {
  3. //        //1. 自定义构建消息
  4. //        Message msg = MessageBuilder.withBody("hello,SpringAMQP".getBytes(StandardCharsets.UTF_8))
  5. //                .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)    //非持久化
  6.                 .setDeliveryMode(MessageDeliveryMode.PERSISTENT)      //持久化
  7. //                .build();
  8.         //2. 发送消息
  9.         rabbitTemplate.convertAndSend("normal.direct", "hi", "hello", new MessagePostProcessor() {
  10.             @Override
  11.             public Message postProcessMessage(Message message) throws AmqpException {
  12.                 message.getMessageProperties().setExpiration("10000");
  13.                 return message;
  14.             }
  15.         });
  16.     }
复制代码
  注意，自定义构建的消息是不具备将消息转化为JSON格式的，而我们之前设置了发送消息是JSON格式的，以是这里不能自定义消息。
  当消息从normal.direct交换机到normal.queue队列，过了10s后，发现还没有人吸收，则会将消息由dlx.direct交换机发送给dlx.queue队列，这样就真正的斲丧者就可以担当耽误了10s后的消息。
4.2 耽误消息插件

这个插件可以将平凡交换机改造为支持耽误消息功能的交换机，当消息投递到交换机后可以暂存一定时间，到期后再投递到队列

4.2.1 声明耽误交换机

基于注解方式：

  1.     @RabbitListener(bindings = @QueueBinding(
  2.             value = @Queue(name = "delay.queue", durable = "true"),
  3.             exchange = @Exchange(name = "delay.direct", delayed = "true",type = ExchangeTypes.DIRECT),
  4.             key = {"hi"}
  5.     ))
  6.     public void listenDelayQueue(String message) {
  7.         log.info("消费者监听到dlx.queue的消息：{}", message);
  9.     }
复制代码
基于@Bean方式

4.2.2 发送耽误消息

发送消息时需要通过消息头x-delay来设置逾期时间：

  1.     @Test
  2.     public void testSendDelayMessageByPlugin() {
  3.         //发送消息
  4.         rabbitTemplate.convertAndSend("delay.direct", "hi", "hello", new MessagePostProcessor() {
  5.             @Override
  6.             public Message postProcessMessage(Message message) throws AmqpException {
  7.                 message.getMessageProperties().setDelay(10000);
  8.                 return message;
  9.             }
  10.         });
  11.     }
复制代码
耽误插件更方便，只要声明一个耽误交换机、队列及其斲丧者即可，不消声明两套，值得注意的是发送消息中耽误函数变成了setDelay
4.3 取消超时订单




免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

立聪堂德州十三局店

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表