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