RabbitMQ知识点

悠悠 2022-10-25 15:28 86阅读 0赞

## 1、什么是 rabbitmq ##

采用 AMQP 高级消息队列协议的一种消息队列技术,最大的特点就是消费并不需要确保提供方存在,实现了服务之间的高度解耦

## 2、为什么要使用 rabbitmq ##

（1）在分布式系统下具备异步,削峰,负载均衡等一系列高级功能;

（2）拥有持久化的机制，进程消息，队列中的信息也可以保存下来。

（3）实现消费者和生产者之间的解耦。

（4）对于高并发场景下，利用消息队列可以使得同步访问变为串行访问达到一定量的限流，利于数据库的操作。

（5）可以使用消息队列达到异步下单的效果，排队中，后台进行逻辑下单。

## 3、使用 rabbitmq 的场景 ##

（1）服务间异步通信

（2）顺序消费

（3）定时任务

（4）请求削峰

## 4、使用了消息队列会有什么缺点? ##

（1）系统可用性降低:你想啊，本来其他系统只要运行好好的，那你的系统就是正常的。现在你非要加个消息队列进去，那消息队列挂了，你的系统不是呵呵了。因此，系统可用性降低

（2）系统复杂性增加:要多考虑很多方面的问题，比如一致性问题、如何保证消息不被重复消费，如何保证保证消息可靠传输。因此，需要考虑的东西更多，系统复杂性增大。  
（3）一致性问题：A 系统处理完了直接返回成功了，人都以为你这个请求就成功了；但是问题是，要是 BCD 三个系统那里，BD 两个系统写库成功了，结果 C 系统写库失败了，咋整？你这数据就不一致了。

## 5、使用 RabbitMQ 有什么好处？ ##

（1）服务间高度解耦：系统A在代码中直接调用系统B和系统C的代码，如果将来D系统接入，系统A还需要修改代码，过于麻烦！

（2）异步通信性能高：将消息写入消息队列，非必要的业务逻辑以异步的方式运行，加快响应速度

（3）流量削峰：对于高并发场景下，利用消息队列可以使得同步访问变为串行访问达到一定量的限流，利于数据库的操作。

## 6、如何确保消息正确地发送至 RabbitMQ？ 如何确保消息接收方消费了消息？ ##

**发送方确认模式**

将信道设置成 confirm 模式（发送方确认模式），则所有在信道上发布的消息都会被指派一个唯一的 ID。

一旦消息被投递到目的队列后，或者消息被写入磁盘后（可持久化的消息），信道会发送一个确认给生产者（包含消息唯一 ID）。

如果 RabbitMQ 发生内部错误从而导致消息丢失，会发送一条 nack（notacknowledged，未确认）消息。

发送方确认模式是异步的，生产者应用程序在等待确认的同时，可以继续发送消息。当确认消息到达生产者应用程序，生产者应用程序的回调方法就会被触发来处理确认消息。

**接收方确认机制**

消费者接收每一条消息后都必须进行确认（消息接收和消息确认是两个不同操作）。只有消费者确认了消息，RabbitMQ 才能安全地把消息从队列中删除。

这里并没有用到超时机制，RabbitMQ 仅通过 Consumer 的连接中断来确认是否需要重新发送消息。也就是说，只要连接不中断，RabbitMQ 给了 Consumer 足够长的时间来处理消息。保证数据的最终一致性；

## 7、如何保证消息不被重复消费？ ##

为什么会出现消息重复？消息重复的原因有两个：1.生产时消息重复，2.消费时消息重复。

**1.1 生产时消息重复**  
由于生产者发送消息给MQ，在MQ确认的时候出现了网络波动，生产者没有收到确认，实际上MQ已经接收到了消息。这时候生产者就会重新发送一遍这条消息。

生产者中如果消息未被确认，或确认失败，我们可以使用定时任务+（redis/db）来进行消息重试。

@Component
    @Slf4J
    public class SendMessage { 
        @Autowired
        private MessageService messageService;
     
        @Autowired
        private RabbitTemplate rabbitTemplate;
     
        // 最大投递次数
        private static final int MAX_TRY_COUNT = 3;
     
        /** * 每30s拉取投递失败的消息, 重新投递 */
        @Scheduled(cron = "0/30 * * * * ?")
        public void resend() { 
            log.info("开始执行定时任务(重新投递消息)");
     
            List<MsgLog> msgLogs = messageService.selectTimeoutMsg();
            msgLogs.forEach(msgLog -> { 
                String msgId = msgLog.getMsgId();
                if (msgLog.getTryCount() >= MAX_TRY_COUNT) { 
                    messageService.updateStatus(msgId, Constant.MsgLogStatus.DELIVER_FAIL);
                    log.info("超过最大重试次数, 消息投递失败, msgId: {}", msgId);
                } else { 
                    messageService.updateTryCount(msgId, msgLog.getNextTryTime());// 投递次数+1
     
                    CorrelationData correlationData = new CorrelationData(msgId);
                    rabbitTemplate.convertAndSend(msgLog.getExchange(), msgLog.getRoutingKey(), MessageHelper.objToMsg(msgLog.getMsg()), correlationData);// 重新投递
     
                    log.info("第 " + (msgLog.getTryCount() + 1) + " 次重新投递消息");
                }
            });
     
            log.info("定时任务执行结束(重新投递消息)");
        }
    }

**1.2消费时消息重复**  
消费者消费成功后，再给MQ确认的时候出现了网络波动，MQ没有接收到确认，为了保证消息被消费，MQ就会继续给消费者投递之前的消息。这时候消费者就接收到了两条一样的消息。  
**配置yml重试策略**

spring:
      rabbitmq:
        listener:
          simple:
            retry:
              enabled: true # 开启消费者进行重试
              max-attempts: 5 # 最大重试次数
              initial-interval: 3000 # 重试时间间隔

**由于重复消息是由于网络原因造成的，因此不可避免重复消息。但是我们需要保证消息的幂等性。**