大家好，我是大堯。

1. 為什么你們公司選擇RabbitMQ作為消息中間件

在消息隊(duì)列選型時(shí)，我們調(diào)研了市場(chǎng)上比較常用ActiveMQ，RabbitMQ，RocketMQ，Kafka。

1. RabbitMQ相對(duì)成熟穩(wěn)定，這是我們選擇它最主要的原因。
2. 社區(qū)比較活躍，有完善的資料可以參考。
3. Rabbitmq的吞吐量可以達(dá)到萬(wàn)級(jí)，完全滿足我們系統(tǒng)的要求。
4. RabbitMQ是Erlang語(yǔ)言開(kāi)發(fā)的，性能比較好。
5. 有完善的可視化界面，方便查看。

2. 消息隊(duì)列的優(yōu)點(diǎn)和缺點(diǎn)有哪些

優(yōu)點(diǎn)有：

• 異步處理 - 相比于傳統(tǒng)的串行、并行方式，提高了系統(tǒng)吞吐量。
• 應(yīng)用解耦 - 系統(tǒng)間通過(guò)消息通信，不用關(guān)心其他系統(tǒng)的處理。
• 流量削鋒 - 可以通過(guò)消息隊(duì)列長(zhǎng)度控制請(qǐng)求量；可以緩解短時(shí)間內(nèi)的高并發(fā)請(qǐng)求。

缺點(diǎn)有：

• 系統(tǒng)可用性降低
• 系統(tǒng)復(fù)雜度提高

3. RabbitMQ常用的工作模式有哪些

3.1 簡(jiǎn)單模型

• p：生成者
• C：消費(fèi)者
• 紅色部分：quene，消息隊(duì)列

3.2 工作模型

這種模式下一條消息只能由一個(gè)消費(fèi)者進(jìn)行消費(fèi)，默認(rèn)情況下，每個(gè)消費(fèi)者是輪詢消費(fèi)的。

• p：生成者
• C1、C2：消費(fèi)者
• 紅色部分：quene，消息隊(duì)列

3.3 發(fā)布訂閱模型（fanout）

這種模型中生產(chǎn)者發(fā)送的消息所有消費(fèi)者都可以消費(fèi)。

• p：生成者
• X：交換機(jī)
• C1、C2：消費(fèi)者
• 紅色部分：quene，消息隊(duì)列

3.4 路由模型（routing）

這種模型消費(fèi)者發(fā)送的消息，不同類型的消息可以由不同的消費(fèi)者去消費(fèi)。

• p：生成者
• X：交換機(jī)，接收到生產(chǎn)者的消息后將消息投遞給與routing key完全匹配的隊(duì)列
• C1、C2：消費(fèi)者
• 紅色部分：quene，消息隊(duì)列

3.5 主題模型（topic）

這種模型和direct模型一樣，都是可以根據(jù)routing key將消息路由到不同的隊(duì)列，只不過(guò)這種模型可以讓隊(duì)列綁定routing key 的時(shí)候使用通配符。這種類型的routing key都是由一個(gè)或多個(gè)單詞組成，多個(gè)單詞之間用.分割。

通配符介紹：

*：只匹配一個(gè)單詞

#：匹配一個(gè)或多個(gè)單詞

4. 如何保證消息不丟失（如何保證消息的可靠性）

一條消息從生產(chǎn)到消費(fèi)經(jīng)歷了三個(gè)階段，分別是生產(chǎn)者，MQ和消費(fèi)者，對(duì)于RabbitMQ來(lái)說(shuō)，消息的傳遞還涉及到交換機(jī)。因此RabbitMQ出現(xiàn)消息丟失的情況有四個(gè)

分別是

1. 消息生產(chǎn)者沒(méi)有成功將消息發(fā)送到MQ導(dǎo)致消息丟失
2. 交換機(jī)未路由到消息隊(duì)列導(dǎo)致消息丟失
3. 消息在MQ中時(shí)，MQ發(fā)生宕機(jī)導(dǎo)致消息丟失
4. 消費(fèi)者消費(fèi)消息時(shí)出現(xiàn)異常導(dǎo)致消息丟失

針對(duì)上面提到的四種情況，分別進(jìn)行處理

1. amqp協(xié)議提供了事務(wù)機(jī)制，在投遞消息時(shí)開(kāi)啟事務(wù)，如果消息投遞失敗，則回滾事務(wù)，很少有人去使用事務(wù)。除了事務(wù)之外，RabbitMQ還提供了生產(chǎn)者確認(rèn)機(jī)制（publisher confirm）。生產(chǎn)者將信道設(shè)置成confirm（確認(rèn)）模式，一旦信道進(jìn)入confirm模式，所有在該信道上面發(fā)布的消息都會(huì)被指派一個(gè)唯一的ID（從1開(kāi)始），一旦消息被投遞到所有匹配的隊(duì)列之后，RabbitMQ就會(huì)發(fā)送一個(gè)確認(rèn)（Basic.Ack）給生產(chǎn)者（包含消息的唯一ID），這就使得生產(chǎn)者知曉消息已經(jīng)正確到達(dá)了目的地了。

# 開(kāi)啟生產(chǎn)者確認(rèn)機(jī)制，
# 注意這里確認(rèn)的是是否到達(dá)交換機(jī)
spring.rabbitmq.publisher-confirm-type=correlated
@RestController
public?class?Producer?{

????@Autowired
????private?RabbitTemplate?rabbitTemplate;

????@GetMapping("send")
????public?void?sendMessage(){
????????/**
?????????*?生產(chǎn)者確認(rèn)消息
?????????*/
????????rabbitTemplate.setConfirmCallback(new?RabbitTemplate.ConfirmCallback()?{
????????????@Override
????????????public?void?confirm(CorrelationData?correlationData,?boolean?ack,?String?cause)?{
????????????????System.out.println(correlationData);
????????????????System.out.println(ack);
????????????????System.out.println(cause);
????????????}
????????});
????????rabbitTemplate.convertAndSend("s","error","這是一條錯(cuò)誤日志?。?！");
????}
}


2. 消息從交換機(jī)未能匹配到隊(duì)列時(shí)將此條消息返回給生產(chǎn)者

spring.rabbitmq.publisher-returns=true
@RestController
public?class?Producer?{

????@Autowired
????private?RabbitTemplate?rabbitTemplate;

????@GetMapping("send")
????public?void?sendMessage(){
????????/**
?????????*?消息未達(dá)隊(duì)列時(shí)返回該條消息
?????????*/
????????rabbitTemplate.setReturnsCallback(new?RabbitTemplate.ReturnsCallback()?{
????????????@Override
????????????public?void?returnedMessage(ReturnedMessage?returnedMessage)?{
????????????????System.out.println(returnedMessage);
????????????}
????????});
????????rabbitTemplate.convertAndSend("s","error","這是一條錯(cuò)誤日志?。?！");
????}
}


3. 消息在交換機(jī)或隊(duì)列中發(fā)生丟失，我們只需要將交換機(jī)和隊(duì)列進(jìn)行持久化。

/**
??*?定義一個(gè)持久化的topic交換機(jī)
??*?durable?持久化
??*?@return
?*/
@Bean
public?Exchange?exchangeJavatrip(){
????return?ExchangeBuilder.topicExchange(EXCHANGE).durable(true).build();
}

/**
?*?定義一個(gè)持久化的隊(duì)列
?*?durable?持久化
?*?@return
?*/
@Bean
public?Queue?queueJavatrip(){
????return?QueueBuilder.durable(QUEUE).build();
}


4. 消費(fèi)者開(kāi)啟手動(dòng)簽收模式，消費(fèi)完成后進(jìn)行ack確認(rèn)。

spring.rabbitmq.listener.simple.acknowledge-mode=manual
@RabbitListener(queues?=?MqConfig.QUEUE)
public?void?receive(String?body,?Message?message,?Channel?channel)?throws?Exception{
????long?deliveryTag?=?message.getMessageProperties().getDeliveryTag();
????System.out.println(deliveryTag);
????//?系統(tǒng)業(yè)務(wù)邏輯判斷是否簽收
????if(deliveryTag?%?2?==?0){
????????channel.basicAck(deliveryTag,false);
????}else{
????????//?第二個(gè)參數(shù)是否批量確認(rèn)，第三個(gè)參數(shù)是否重新回隊(duì)列
????????channel.basicNack(deliveryTag,false,true);
????}
}


5. 如何保證消息不重復(fù)消費(fèi)（如何保證消息的冪等性）

消息重復(fù)的原因有兩個(gè)：

1. 生產(chǎn)時(shí)消息重復(fù)
   由于生產(chǎn)者發(fā)送消息給MQ，在MQ確認(rèn)的時(shí)候出現(xiàn)了網(wǎng)絡(luò)波動(dòng)，生產(chǎn)者沒(méi)有收到確認(rèn)，實(shí)際上MQ已經(jīng)接收到了消息。這時(shí)候生產(chǎn)者就會(huì)重新發(fā)送一遍這條消息。

2. 消費(fèi)時(shí)消息重復(fù)。
   消費(fèi)者消費(fèi)成功后，在給MQ確認(rèn)的時(shí)候出現(xiàn)了網(wǎng)絡(luò)波動(dòng)，MQ沒(méi)有接收到確認(rèn)，為了保證消息被消費(fèi)，MQ就會(huì)繼續(xù)給消費(fèi)者投遞之前的消息。這時(shí)候消費(fèi)者就接收到了兩條一樣的消息。

由于消息重復(fù)是網(wǎng)絡(luò)波動(dòng)等原因造成的，無(wú)法避免，我們能做的的就是保證消息的冪等性，以防業(yè)務(wù)重復(fù)處理。具體處理方案為：

讓每個(gè)消息攜帶一個(gè)全局的唯一ID，即可保證消息的冪等性，具體消費(fèi)過(guò)程為：

1. 消費(fèi)者獲取到消息后先根據(jù)id去查詢r(jià)edis/db是否存在該消息。
2. 如果不存在，則正常消費(fèi)，消費(fèi)完畢后寫(xiě)入redis/db。
3. 如果存在，則證明消息被消費(fèi)過(guò)，直接丟棄。

@RabbitListener(queues?=?MqConfig.QUEUE)
public?void?receive(Message?message,?Channel?channel){

????String?messageId?=?message.getMessageProperties().getMessageId();
????String?body?=?new?String(message.getBody());
????String?redisId?=?redisTemplate.opsForValue().get(messageId) "";
????//?如果redis中存有當(dāng)前消息的消息id
????//?則證明消費(fèi)過(guò)
????if(messageId.equals(redisId)){
????????return;
????}
????redisTemplate.opsForValue().set(messageId,?UUID.randomUUID());
}


6. 消息大量堆積應(yīng)該怎么處理

消息堆積的原因有兩個(gè)

1. 網(wǎng)絡(luò)故障，消費(fèi)者無(wú)法正常消費(fèi)
2. 消費(fèi)方消費(fèi)后未進(jìn)行ack確認(rèn)

解決方案如下：

1. 檢查并修復(fù)消費(fèi)者故障，使其正常消費(fèi)
2. 編寫(xiě)臨時(shí)程序?qū)⒍逊e的消息發(fā)送到容量更大的MQ集群，增加消費(fèi)者快速消費(fèi)
3. 堆積消息消費(fèi)完畢后，停止臨時(shí)程序，恢復(fù)正常消費(fèi)

7. 死信是什么？死信如何處理

當(dāng)一條消息在隊(duì)列中出現(xiàn)以下三種情況的時(shí)候，該消息就會(huì)變成一條死信。

• 消息被拒絕(basic.reject / basic.nack)，并且requeue = false
• 消息TTL過(guò)期
• 隊(duì)列達(dá)到最大長(zhǎng)度

當(dāng)消息在一個(gè)隊(duì)列中變成一個(gè)死信之后，如果配置了死信隊(duì)列，它將被重新publish到死信交換機(jī)，死信交換機(jī)將死信投遞到一個(gè)隊(duì)列上，這個(gè)隊(duì)列就是死信隊(duì)列。

一條消息成為死信后，一般會(huì)通過(guò)死信隊(duì)列進(jìn)行存庫(kù)，然后定時(shí)將庫(kù)中的死信進(jìn)行重新投遞到消息隊(duì)列上。

8. 如果我有一筆訂單，30分鐘未支付則關(guān)閉訂單，使用RabbitMQ如何來(lái)實(shí)現(xiàn)

RabbitMQ可以使用死信隊(duì)列來(lái)實(shí)現(xiàn)延時(shí)消費(fèi)，用戶下單之后，將訂單信息投遞到消息隊(duì)列中，并且設(shè)置消息過(guò)期時(shí)常為30分鐘。如果用戶支付則正常關(guān)閉訂單，如果用戶未支付，消息達(dá)到過(guò)期時(shí)間，消息會(huì)進(jìn)入死信交換，由消費(fèi)者進(jìn)行消費(fèi)死信隊(duì)列來(lái)關(guān)閉訂單。

9. RabbitMQ如何保證高可用

RabbitMQ有兩種集群模式，分別是普通集群和鏡像集群，普通模式無(wú)法保證RabbitMQ的高可用。

普通集群

假如有三個(gè)節(jié)點(diǎn)，rabbitmq1、rabbitmq2、rabbitmq3，消息實(shí)際上只存在于其中一個(gè)節(jié)點(diǎn)，三個(gè)節(jié)點(diǎn)僅有相同的元數(shù)據(jù)，即隊(duì)列的結(jié)構(gòu)，當(dāng)消息進(jìn)入rabbitmq2節(jié)點(diǎn)的queue后，consumer從rabbitmq1的節(jié)點(diǎn)進(jìn)行消費(fèi)，rabbitmq1和rabbitmq2會(huì)進(jìn)行臨時(shí)通信，從rabbitmq2中獲取消息然后返回給consumer。

這種模式存在以下兩個(gè)問(wèn)題：

1. 當(dāng)rabbitmq2宕機(jī)后，消息無(wú)法正常消費(fèi)，沒(méi)有做到真正的高可用

2. 實(shí)際數(shù)據(jù)還是在單個(gè)實(shí)例上，存在瓶頸問(wèn)題

鏡像集群

假如有三個(gè)節(jié)點(diǎn)，rabbitmq1、rabbitmq2、rabbitmq3，每個(gè)實(shí)例之間都可以相互通信，每次生產(chǎn)者寫(xiě)消息到queue的時(shí)候，每個(gè)rabbitmq節(jié)點(diǎn)上都有queue的消息數(shù)據(jù)和元數(shù)據(jù)。這種模式適用于可靠性要求較高的場(chǎng)景。