前言

MQ（Message Queue）就是消息隊列，其有點有很多：解耦、異步、削峰等等，本文來聊一下RabbitMQ的一些概念以及使用。

RabbitMq

案例

Springboot整合RabbitMQ簡單案例

基本概念

• Exchange：消息交換機，它指定消息按什么規則，路由到哪個隊列。
• Queue：消息隊列載體，每個消息都會被投入到一個或多個隊列。
• Binding：綁定，它的作用就是把exchange和queue按照路由規則綁定起來。
• Routing Key：路由關鍵字，exchange根據這個關鍵字進行消息投遞。
• Producer：消息生產者，就是投遞消息的程序。
• Consumer：消息消費者，就是接受消息的程序。

發布消息到RabbitMQ需要經過兩步：

1. producer → exchange
2. exchange 根據 exchange 的類型和 routing key 確定將消息投遞到哪個隊列

工作流程

了解了RabbitMQ的一些概念，我們來捋捋使用RabbitMQ的流程：

1. 創建Exchange
2. 創建Queue
3. 將Queue綁定進Exchange中（此處會設置routing key）
4. 生產者發布消息
5. 消費者訂閱消息

交換機（Exchange）

交換機可以綁定隊列，綁定時可以給隊列指定路由（Routing key）和參數（Arguments）

所有的消息發送都是經過交換機轉發到隊列的，而不是直接到隊列中

交換機類型：

• direct
• 根據確定的路由(routing key)轉發消息到隊列中（一條消息可以發到多個隊列，只要路由相同）
• fanout
• 路由無效，只要和該交換機綁定的隊列，都能接收到消息
• topic
• 允許路由使用*和#來進行模糊匹配
• *表示一個單詞
• 表示任意數量(零個或多個)單詞
• 例如：如果隊列的路由為com.# 那么往交換機發消息是，路由填com.ccc 隊列就可以收到消息
• headers
• 忽略路由，由參數（Arguments）來確定轉發的隊列

消息過期時間TTL

有兩種方式設置TTL，創建隊列時設置整個隊列的TTL或者在發送消息時單獨設置每條消息的TTL，消息存活時間取兩者的最小值。

1. 創建隊列時設置
2. 是消息的存活時間，不是隊列的存活時間，別搞混了。
3. @Beanpublic Queue queue(){ Map<String, Object> args = new HashMap<>(); args.put("x-message-ttl", 5000); // 設置隊列中的消息5秒過期 return new Queue("queueName",true, false, false, args);}
4. 發送消息時設置
5. public void makeOrder(String userid,String productid,int num){ String exchangeName = "ttl_exchange"; String routingKey = "ttlmessage"; //給消息設置過期時間 MessagePostProcessor messagePostProcessor = new MessagePostProcessor(){ public Message postProcessMessage(Message message){ // 設置消息5秒過期 message.getMessageProperties().setExpiration("5000"); return message; } } rabbitTemplate.convertAndSend(exchangeName,routingKey,"message",messagePostProcessor);}

死信隊列

死信隊列也是一個正常隊列，只是當綁定了死信隊列的隊列滿足相應條件，就會將滿足條件的消息轉移到死信隊列中。

進入死信隊列的條件：

1. 消息被拒絕
2. 消息過期（超時）
3. 隊列達到最大長度

死信隊列的配置：

1. 按照正常步驟定義一個隊列（交換機、隊列、綁定）
2. 給需要綁定死信隊列的隊列添加x-dead-letter-exchange（死信隊列的交換機）和x-dead-letter-routing-key（死信隊列的路由）參數
3. @Beanpublic Queue queue(){ Map<String, Object> args = new HashMap<>(); args.put("x-dead-letter-exchange", "死信隊列交換機名稱"); args.put("x-dead-letter-routing-key", "死信隊列路由"); return new Queue("queueName",true, false, false, args);}

如何保證MQ消息正確送達與消費

可靠性生產和推送

步驟：

1. 發送消息前數據庫保存MQ消息發送日志
2. MQ消息發送后使用回調更新日志狀態

實現：

上面我們講了，發布消息到RabbitMQ需要經過兩步：

producer → exchange
exchange 根據 exchange 的類型和 routing key 確定將消息投遞到哪個隊列

所以，發布消息的確認也分兩個部分，以下是確認步驟：

1. 修改MQ應答機制（yml）
2. spring: rabbitmq: username: rmq password: 123456 virtual-host: / host: localhost port: 5672 # 發送消息確認，producer -> exchange publisher-confirm-type: correlated # 發送消息確認，exchange -> queue publisher-returns: true
3. 新增mq的回調方法
4. /** * PostConstruct注解好多人以為是Spring提供的。其實是Java自己的注解。 * Java中該注解的說明：@PostConstruct該注解被用來修飾一個非靜態的void（）方法。 * 被@PostConstruct修飾的方法會在服務器加載Servlet的時候運行，并且只會被服務器執行一次。 * PostConstruct在構造函數之后執行，init（）方法之前執行。 * Constructor(構造方法) -> @Autowired(依賴注入) -> @PostConstruct(注釋的方法) */@PostConstructprivate void regCallBack() { // producer -> exchange 成功或失敗都會觸發此回調 rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { // 這個id是在消息發送的時候傳入的 String id = correlationData.getId(); // 如果ack為true代表消息被mq成功接收 if (!ack) { // 應答失敗，修改日志狀態 System.out.println("exchange 應答失敗，做失敗處理！"); } else { // 應答成功，修改日志狀態 System.out.println("exchange 成功處理"); } } }); // 這個回調只有exchange -> queue 失敗時才會觸發 rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() { @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { System.out.println("exchange -> queue 發送失敗"); } });}
5. 修改MQ發送消息的方法，增加日志id的傳遞
6. String correlationId = "這是日志id";rabbitTemplate.convertAndSend(exchange, routeKey, message, new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { // 消費者需要correlationId才做這個處理 message.getMessageProperties().setCorrelationId(correlationId); return message; }}, new CorrelationData(correlationId));// 如果消費者不需要獲取correlationId，則用下面這種即可rabbitTemplate.convertAndSend(exchange, routeKey, msg, new CorrelationData(correlationId));

可靠性消費

步驟：

• 開啟手動應答
• 監聽器增加手動應答邏輯

實現：

1. 開啟手動應答
2. spring: rabbitmq: username: rmq password: 123456 virtual-host: / host: localhost port: 5672 listener: simple: acknowledge-mode: manual # 將自動應答ack模式改成手動應答
3. acknowledge-mode有三種類型：
4. nome：不進行ack，rabbitmq默認消費者正確處理所有請求
5. munual：手動確認
6. auto：自動確認消息（默認類型）。如果消費者拋出異常，則消息重回隊列。
7. 監聽器增加手動應答邏輯
8. @RabbitListener(queues = {"隊列名字"})public void messageConsumer(String orderMsg, Channel channel, @Headers Map<String,Object> headers) throws Exception{ // 需要producer做相應處理，consumer才能拿到correlationId String correlationId = messages.getMessageProperties().getCorrelationId(); System.out.println("消息為：" + orderMsg); long tag = Long.parseLong(headers.get(AmqpHeaders.DELIVERY_TAG).toString()); try { // 消費成功，進行確認 channel.basicAck(tag, false); } catch (IOException e) { // 消費失敗，重發 // requeue代表是否重發，為false則直接將消息丟棄，有死信就進入死信隊列 channel.basicNack(tag, false, true); }}

總結

本文介紹了RabbitMQ的一些概念和簡單使用，有不少東西其實是沒有講清楚的，比如publisher-confirm-type和acknowledge-mode的幾種類型的區別等等。主要是在官方文檔找不到相關的細致描述，查文檔的能力還是有待提高。。。