6月iOS設備好評榜：第一蟬聯榜首近一年
作為安兔兔各種榜單里變化最小的那個，2023年6月的iOS好評榜和上個月相比沒有任何排名上的變化，僅僅是部分設備好評率的下降，長年累月的用戶評價和逐漸退出市場的老款機器讓這
本篇文章不再介紹RabbitMQ具體實現原理，直接介紹如何保證消息的可靠性問題。所謂可靠性，指消息不重不漏。
圖片
??生產者-消費者模型用于描述兩類進程（生產者和消費者）之間的數據交互。可以被認為是獨立的服務，生產者負責生成數據，消費者負責處理這些數據。在分布式系統中，隊列在其中扮演了消息（數據）傳遞的功能。
圖片
關于消息隊列的作用，一般解讀為：
解耦：生產者和消費者獨立運作，無需知道對方的運行狀態。
異步：并非實時，生產者不必關注消費端的消費情況。
削峰：限制流量，防止消費者過載。
??這其實不難理解，就像生活中下單-快遞-簽收的過程。這個過程和上邊的生產者-消費者模型恰有異曲同工之妙。
圖片
這個過程中，
如果包裹被粗略的認為是一條消息，那么快件在郵寄過程中丟失了，就是消息丟失。快件從發貨到簽收，我們不用去關心中間發生了什么。但是要是沒收到貨，那得給我個理由。
??就上邊的快件丟失問題，怎么知道快遞為何沒有收到？很簡單，一段一段的排查：
相應的，如果生產環境中突然發現諸如：告警、服務宕機、數據流轉異常等問題時，我們也會在鏈路上（A、B、C三處）逐一排查。
為確保消息從生產端可靠地投遞到RabbitMQ，我們需要考慮以下幾個關鍵點：
網絡故障：消息可能在傳輸過程中因網絡問題而丟失。
RabbitMQ故障：如果RabbitMQ宕機，消息也可能丟失。
圖片
對應解決方案：
事務在RabbitMQ中可能會影響性能，因為它們需要在所有節點上同步狀態。因此，RabbitMQ盡量避免使用事務。核心代碼：
private static void executeTransaction(Channel channel) throws IOException { boolean transactionSuccess = false; try { // 開啟事務 channel.txSelect(); // 執行一系列消息操作，例如：channel.basicPublish(exchange, routingKey, message); // 提交事務 channel.txCommit(); transactionSuccess = true; } catch (ShutdownSignalException | IOException e) { // 回滾事務 if (!transactionSuccess) { channel.txRollback(); } throw e; } }發布者確認機制允許發布者知道消息是否已經被RabbitMQ成功接收：
public static void sendPersistentMessage(String host, String queueName, String message) { try (Connection connection = new ConnectionFactory().setHost(host).newConnection(); Channel channel = connection.createChannel()) { // 啟用發布者確認 channel.confirmSelect(); // 將消息設置為持久化 AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder() .deliveryMode(2) .build(); // 添加確認監聽器 channel.addConfirmListener(new ConfirmListener() { @Override public void handleAck(long deliveryTag, boolean multiple) throws IOException { System.out.println("消息已確認: " + deliveryTag); // 消息正確到達Broker時的處理邏輯 } @Override public void handleNack(long deliveryTag, boolean multiple) throws IOException { System.out.println("消息未確認: " + deliveryTag); // 因為內部錯誤導致消息丟失時的處理邏輯 } }); channel.basicPublish("", queueName, properties, message.getBytes()); // 等待消息確認，或者超時 boolean allConfirmed = channel.waitForConfirms(); if (allConfirmed) { //所有消息都已確認 } else { //超時或其它 } } catch (IOException | TimeoutException | InterruptedException e) { e.printStackTrace(); }}在RabbitMQ中，消息的持久化它確保消息不僅存儲在內存中，而且也安全地保存在磁盤上。這樣，即使在RabbitMQ服務崩潰或重啟的情況下，消息也不會丟失，可以從磁盤恢復。
圖片
消息到達RabbitMQ后通過Exchange交換機，路由給queue隊列，最后發送給消費端。
圖片
從RabbitMQ設計上看，消息的持久化應該從以下方面入手：
// 設置 durable = true; channel.exchangeDeclare(exchangeName, "direct", durable);// 設置 MessageProperties.PERSISTENT_TEXT_PLAINchannel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());//設置 boolean durable = true;channel.queueDeclare(queueName, durable, exclusive, false, null);這樣，如果RabbitMQ收到消息后掛了，重啟后會自行從磁盤上恢復消息。
如果上述生產端、消息隊列都正確投遞，那么問題出現在消費端是否可以正確消費？
圖片
消費者在成功處理了一條消息后通知RabbitMQ，這樣RabbitMQ在收到確認后才會移除隊列中的消息。
默認情況下，以下3種原因導致消息丟失：
1、 網絡故障：消費端還沒接收到消息之前，發生網絡故障導致消息丟失；
2、 未接收消息前服務宕機：消費端突然掛機未接收到消息，此時消息會丟失；
3、 處理過程中服務宕機：消費端正確接收到消息，但在處理消息的過程中發生異常或宕機了，消息也會丟失。
這是因為RabbitMQ的自動ack機制，即默認RabbitMQ在消息發出后，不管消費端是否接收到，是否處理完，就立即刪除這條消息，導致消息丟失。
圖片
應對方案：
DeliverCallback deliverCallback = (consumerTag, delivery) -> { try { //接收消息，業務處理 //設置手動確認 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } catch (Exception e) { //發生異常時，可以選擇重新發送消息或進行錯誤處理 // 例如，可以選擇負確認（nack），讓消息重回隊列 // channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); }};//設置autoAck為false，表示關閉自動確認機制，改為手動確認channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {});以上3種解決辦法理論上可靠，但是系統的異常或者故障比較偶然，我們沒法做到100%消息不丟失。因此需要介入補償機制或者人工干預。這是我們的最后一道防線。
如何做消息補償呢？其實就是將消息入庫，通過定時任務重新發送失敗的消息。詳細流程如下：
圖片
標注：
超過最大失敗次數后，對于無法被正常消費的消息可移入死信隊列。
以上，我們知道了消息丟失問題如何處理？那么對于消息重復的問題，下面做個介紹。
消息重復消費是指在消息隊列中，同一條消息被不同的消費者多次消費處理。
產生原因：
對應解決方案：
設計消費者的消息處理邏輯時，要保證即使消息被多次消費，也不會對系統狀態產生不良影響。冪等性可以通過以下方式實現：
使用唯一標識符（如訂單號、massageID）來識別消息，并在消費者中實現去重邏輯：
通過手動確認消息，控制消息何時從隊列中移除：
代碼演示:
消費者端去重邏輯
@RabbitListener(queues = "queueName", acknowledgeMode = "MANUAL")public void receiveMessage(Message message, Channel channel) throws IOException { String messageId = message.getMessageProperties().getMessageId(); // 檢查消息是否已消費 if (messageAlreadyProcessed(messageId)) { // 消息已消費，確認消息并返回 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); return; } // 處理消息 try { processMessage(message); // 消息處理成功，持久化消息ID并確認消息 persistMessageId(messageId); channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } catch (Exception e) { // 處理失敗，可以選擇重新入隊或丟棄 boolean requeue = shouldRequeue(message); channel.basicReject(message.getMessageProperties().getDeliveryTag(), requeue); }}生產者端發布確認
void sendWithConfirm(AmqpTemplate amqpTemplate, Message message) throws IOException { ConfirmCallback confirmCallback = (correlationData, ack, cause) -> { if (!ack) { // 處理消息發送失敗的邏輯 // ... } }; amqpTemplate.setConfirmCallback(confirmCallback); amqpTemplate.convertAndSend("exchangeName", "routingKey", message);}具體實現需要根據實際業務邏輯和RabbitMQ配置進行調整。
以上介紹了RabbitMQ保證消息可靠性的問題、產生原因、解決方案等。不足之處，歡迎指正。
本文鏈接：http://www.tebozhan.com/showinfo-26-87491-0.htmlRabbitMQ如何保證消息可靠性？
聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
下一篇： 純 CSS 實現標簽自動顯示超出數量
Copyright ? 2016-2023 天津谷騏科技有限公司 版權所有 sitemap.xml
違法及侵權請聯系：2376512515@qq.com 津ICP備18001702號
津公網安備 12010102000574號