RabbitMQ相关问题

Mybatis框架相关问题

RabbitMQ相关问题

一、RabbitMQ的核心组件和工作原理？
二、如何保证消息可靠投递不丢失的？
三、RabbitMQ如何保证消息的幂等性？
四、什么是死信队列？死信队列是如何导致的？
五、RabbitMQ死信队列是如何导致的？
六、什么是延迟队列？RabbitMQ 如何实现延迟队列？
七、RabbitMQ的高可用机制有了解嘛？
八、如果有百万消息堆积在MQ中，如何解决？
九、如何解决RabbitMQ中因为消息堆积而导致的消息过期失效的问题？

一、RabbitMQ的核心组件和工作原理？

在这里插入图片描述
交换机有四类：
1、Fanout Exchange

投递到所有绑定的队列，不需要规则，不需要匹配，相当于广播、群发；
在这里插入图片描述
2、Direct Exchange

根据路由键精确匹配进行路由消息队列；
在这里插入图片描述
3、Topic Exchange

通配符匹配，相当于模糊匹配；

# 匹配多个单词，* 匹配一个单词，用 . 隔开的为一个单词：

在这里插入图片描述
4、Headers Exchange

基于消息内容中的headers属性进行匹配；
在这里插入图片描述

二、如何保证消息可靠投递不丢失的？

在这里插入图片描述
要确保一下四个步骤全部成功；
① 代表消息从生产者发送到Exchange;
② 代表消息从Exchange路由到Queue；
③ 代表消息在Queue中存储；
④ 代表消费者监听Queue并消费消息；

① 代表消息从生产者发送到Exchange;
- 方案：开启Confirm确认模式；
- 消息未投递成功，采取补偿措施，或是记录日志，或是发送通知让负责人知道；
```
// 实现接口
implements RabbitTemplate.ConfirmCallback// 重写方法
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause)
```

② 代表消息从Exchange路由到Queue；

方案：开启Return返回模式；
消息未投递成功，采取补偿措施，或是记录日志，或是发送通知让负责人知道；

// 实现接口
implements RabbitTemplate.ReturnsCallback/*** 消息从 交换机 --> 到 --> 队列，如果失败了，就会回调该方法* （失败了才触发该方法，成功是不会触发该方法的）** 比如说磁盘满** @param returned*/
@Override
public void returnedMessage(ReturnedMessage returned)

③ 代表消息在Queue中存储；

方案：
1、交换机持久化

    /*** 声明创建一个FanoutExchange交换机** @return*/@Beanpublic DirectExchange directExchange() {return ExchangeBuilder.directExchange(DIRECT_EXCHANGE_NAME).durable(true) // 持久化.build();}

2、队列持久化

    /*** 声明创建一个队列** @return*/@Beanpublic Queue directQueue() {return QueueBuilder.durable(DIRECT_QUEUE_NAME).build();}

3、消息持久化

	//消息体Message message = MessageBuilder.withBody(json.getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT) //消息持久化.build();

4、集群高可用部署

④ 代表消费者监听Queue并消费消息；

方案：手动确认消息

@Slf4j
@Component
public class MyRabbitListener {@RabbitListener(queues = {RabbitConfig.DIRECT_QUEUE_NAME})public void onMessage(String msg, @Headers Map<String, Object> header, Message message, Channel channel) {try {System.out.println("[RabbitListener]接收到的消息: " + msg);//处理业务//业务处理成功，手动确认消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);//业务处理失效，不确认消息，并且重新入队，这样又可以重新消费// channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);//拒绝消息// channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {e.printStackTrace();}}
}