2024年大数据最全RabbitMQ学习总结-消息的可靠性(1)，一线大厂架构师都推荐的大数据开发零基础大全

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，涵盖了95%以上大数据知识点，真正体系化！

由于文件比较多，这里只是将部分目录截图出来，全套包含大厂面经、学习笔记、源码讲义、实战项目、大纲路线、讲解视频，并且后续会持续更新

需要这份系统化资料的朋友，可以戳这里获取

保证MQ消息的可靠性，主要从三个方面：发送者确认可靠性，MQ确认可靠性，消费者确认可靠性。

1.发送者可靠性：主要依赖于发送者重试机制，发送者确认机制；

发送者重试机制，其实就是配置文件配置重试规则，当消息发送失败后，会根据配置的重试次数，进行多次发送重试，如代码：

spring:
  rabbitmq:
    connection-timeout: 1s # 设置MQ的连接超时时间
    template:
      retry:
        enabled: true # 开启超时重试机制
        initial-interval: 1000ms # 失败后的初始等待时间
        multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = initial-interval * multiplier
        max-attempts: 3 # 最大重试次数
1
2
3
4
5
6
7
8
9

发送者确认机制：则是依赖于消息的回执，这其中包括发送者回执，和消费者回执两种，但是这种回执都比较耗性能，会导致消息消费的很慢。并且，这也是需要在配置文件中做配置的：

spring:
  rabbitmq:
    publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型
    publisher-returns: true # 开启publisher return机制
1
2
3
4

并且还要有代码的实现，这种方式极大的影响了性能，：

@Slf4j
@AllArgsConstructor
@Configuration
public class MqConfig {
    private final RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init(){
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                log.error("触发return callback,");
                log.debug("exchange: {}", returned.getExchange());
                log.debug("routingKey: {}", returned.getRoutingKey());
                log.debug("message: {}", returned.getMessage());
                log.debug("replyCode: {}", returned.getReplyCode());
                log.debug("replyText: {}", returned.getReplyText());
            }
        });
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

@Test
void testPublisherConfirm() {
    // 1.创建CorrelationData
    CorrelationData cd = new CorrelationData();
    // 2.给Future添加ConfirmCallback
    cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
        @Override
        public void onFailure(Throwable ex) {
            // 2.1.Future发生异常时的处理逻辑，基本不会触发
            log.error("send message fail", ex);
        }
        @Override
        public void onSuccess(CorrelationData.Confirm result) {
            // 2.2.Future接收到回执的处理逻辑，参数中的result就是回执内容
            if(result.isAck()){ // result.isAck()，boolean类型，true代表ack回执，false 代表 nack回执
                log.debug("发送消息成功，收到 ack!");
            }else{ // result.getReason()，String类型，返回nack时的异常描述
                log.error("发送消息失败，收到 nack, reason : {}", result.getReason());
            }
        }
    });
    // 3.发送消息
    rabbitTemplate.convertAndSend("hmall.direct", "q", "hello", cd);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

2.MQ自身的可靠性：交换机/队列/消息都实现持久化，消息不会丢失，如果是在项目中通过代码创建的交换机/队列/消息，spring默认就是持久化的，如果在mq的客户端手工配置，那就要选定各个参数了。持久化后的消息会直接进入磁盘，不在经过内存了，正常来讲有IO的操作会慢才对，但是在实际的操作中却是非常快。

MQ队列最怕的就是消息积压，导致内存溢出。在3.12版本以后，MQ直接默认就是Laz懒惰队列的模式了，这个模式会直接加载到磁盘，当用到消息的时候，会从磁盘加载到内存，磁盘空间很大，支持数百万级别的存储，所以内存溢出的可能性就会大大降低。我们可以在mq客户端手动设置为lazy队列，也可以在代码中直接实现，代码如下：

@RabbitListener(queuesToDeclare = @Queue(
        name = "lazy.queue",
        durable = "true",
        arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listenLazyQueue(String msg){
    log.info("接收到 lazy.queue的消息：{}", msg);
}
1
2
3
4
5
6
7
8

3.消费者的可靠性：

3.1消费者消费消息后，向MQ发送回执，让MQ知道消息是否正常被消费了，目前回执有三种：

ack：成功处理了消息，MQ从队列中就会删除消息，正常。

nack：失败处理了消息，MQ需要再次投递消息，这会出现一直重试的问题。

reject：消息失败，并拒绝了消息，并且从队列中删除了消息。这个消息被删除了，岂不是数据就丢失了。

对于以上三种回执，基本回执都是固定的，AMQP提供了消息确认的方式，不用写代码，配置就可以,配置有三种：none-配置它失败了，消息会被删除，auto-失败了，消息会回到MQ重新投递，不会丢失，不会被删除，manual-太麻烦，算了。

不过，对于auto的配置，对于返回的异常，会有两种判断：1，如果是业务异常，会自动返回nack

如果是消息处理或者校验异常，会直接进行reject

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。

需要这份系统化资料的朋友，可以戳这里获取

一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！

net/forums/4f45ff00ff254613a03fab5e56a57acb)**

一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！