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)
1
2
3
4
5
6
```

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

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

// 实现接口
implements RabbitTemplate.ReturnsCallback

/**
 * 消息从 交换机 --> 到 --> 队列，如果失败了，就会回调该方法
 * （失败了才触发该方法，成功是不会触发该方法的）
 *
 * 比如说磁盘满
 *
 * @param returned
 */
@Override
public void returnedMessage(ReturnedMessage returned)
1
2
3
4
5
6
7
8
9
10
11
12
13

③ 代表消息在Queue中存储；

方案：
1、交换机持久化

    /**
     * 声明创建一个FanoutExchange交换机
     *
     * @return
     */
    @Bean
    public DirectExchange directExchange() {
        return ExchangeBuilder.directExchange(DIRECT_EXCHANGE_NAME)
                .durable(true) // 持久化
                .build();
    }
1
2
3
4
5
6
7
8
9
10
11

2、队列持久化

    /**
     * 声明创建一个队列
     *
     * @return
     */
    @Bean
    public Queue directQueue() {
        return QueueBuilder.durable(DIRECT_QUEUE_NAME).build();
    }
1
2
3
4
5
6
7
8
9

3、消息持久化

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

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();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24