我有以下 rabbitMq 消費(fèi)者：Consumer consumer = new DefaultConsumer(channel) { @Override
public void handleDelivery(String consumerTag, Envelope envelope, MQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "UTF-8");
sendNotificationIntoTopic(message);
saveIntoDatabase(message);
}
};可能會(huì)出現(xiàn)以下情況：消息已成功發(fā)送到主題與數(shù)據(jù)庫(kù)的連接丟失，因此數(shù)據(jù)庫(kù)插入失敗。結(jié)果我們有數(shù)據(jù)不一致。預(yù)期結(jié)果要么兩個(gè)操作都成功執(zhí)行，要么根本沒(méi)有執(zhí)行。任何解決方案我怎樣才能實(shí)現(xiàn)它？聚苯乙烯目前我有以下想法（請(qǐng)?jiān)u論）我們可以假設(shè)代理不會(huì)丟失任何消息。我們必須訂閱要發(fā)送的主題。將條目保存到數(shù)據(jù)庫(kù)中并設(shè)置status值為“pending”的 字段嘗試向主題發(fā)送數(shù)據(jù)。如果發(fā)送成功 - 更新status值為“成功”的字段我們必須有一個(gè)計(jì)劃作業(yè)，它必須檢查具有掛起狀態(tài)的行。目前可能有兩種情況：3.1 根本沒(méi)有發(fā)送通知3.2 發(fā)送了通知但存入數(shù)據(jù)庫(kù)失?。ǜ怕屎艿偷锌赡埽┧晕覀儽仨氁阅撤N方式區(qū)分這兩種情況：我們可以將來(lái)自主題的消息存儲(chǔ)在集合中，作業(yè)可以檢查消息是否被接受。因此，如果作業(yè)找到與數(shù)據(jù)庫(kù)行對(duì)應(yīng)的消息，我們必須將狀態(tài)更新為“成功”。否則我們必須從數(shù)據(jù)庫(kù)中刪除條目。我認(rèn)為我的想法有一些弱點(diǎn)（例如，如果我們有多節(jié)點(diǎn)應(yīng)用程序，我們必須將消息存儲(chǔ)在 hazelcast（或類(lèi)似物）中，但這是假設(shè)失敗的額外點(diǎn)）
3 回答
慕哥9229398
TA貢獻(xiàn)1877條經(jīng)驗(yàn) 獲得超6個(gè)贊
這是嘗試取消確認(rèn)模式 https://servicecomb.apache.org/docs/distributed_saga_3/的示例 ，它應(yīng)該能夠處理您的問(wèn)題。您應(yīng)該容忍通過(guò)隊(duì)列重復(fù)提交數(shù)據(jù)的機(jī)會(huì)。這是一個(gè)例子：
定義抽象操作并為操作分配 ID 和時(shí)間戳。
將狀態(tài)Pending寫(xiě)入數(shù)據(jù)庫(kù)（可以和1一樣的步驟）
編寫(xiě)一個(gè)偵聽(tīng)器，輪詢(xún)數(shù)據(jù)庫(kù)中所有狀態(tài)為掛起且早于“超時(shí)”的操作
對(duì)于每個(gè)掛起的操作，通過(guò)具有分配 ID 的隊(duì)列發(fā)送數(shù)據(jù)。
接收方應(yīng)該知道 ID，如果 ID 已被處理，則不會(huì)發(fā)生任何事情。
6A。如果您需要 100% 確認(rèn)操作已完成，您需要第二個(gè)隊(duì)列，接收方將在其中發(fā)布消息 ID - DONE。如果不需要這種一致性，請(qǐng)?zhí)^(guò)此步驟。或者，它可以發(fā)布 ID -Failed 失敗原因。
6B。提交方要么等待來(lái)自 6A 的消息，要么通過(guò)將狀態(tài) DONE 寫(xiě)入數(shù)據(jù)庫(kù)來(lái)完成操作。
一旦 sertine 超時(shí)已過(guò)或某個(gè)重試限制已過(guò)。您將狀態(tài)寫(xiě)入操作 FAIL。
您可以通過(guò) ID 回滾將消息發(fā)送到接收方操作。
請(qǐng)注意，所有這些步驟都不涉及技術(shù)事務(wù)。您可以使用非事務(wù)性數(shù)據(jù)庫(kù)執(zhí)行此操作。
我寫(xiě)的是嘗試取消確認(rèn)模式的變體，其中每個(gè)消息接收者都應(yīng)該知道如何管理自己的數(shù)據(jù)。
MM們
TA貢獻(xiàn)1886條經(jīng)驗(yàn) 獲得超2個(gè)贊
如果有足夠的時(shí)間來(lái)修改設(shè)計(jì)，建議使用類(lèi)似 JTA 的 API 來(lái)管理 2phase 提交。甚至 weblogic 和 WebSphere 也支持用于兩階段提交的 XA 資源。
如果時(shí)間線較少，建議執(zhí)行以下操作以減少失敗間隔。
發(fā)送數(shù)據(jù)主題（不提交）（incase topic down, retry to be perform with a interval）
將數(shù)據(jù)寫(xiě)入數(shù)據(jù)庫(kù)
提交數(shù)據(jù)庫(kù)
提交主題
只有當(dāng)?shù)?4 步失敗時(shí)才會(huì)發(fā)生這里失敗。這將導(dǎo)致再次發(fā)送相同的消息。所以接收系統(tǒng)會(huì)收到重復(fù)的消息。在JMS2.0 結(jié)構(gòu)中，每條消息都有唯一的messageID 和CorrelationID。所以找到重復(fù)項(xiàng)有點(diǎn)直截了當(dāng)（但這將在接收系統(tǒng)中處理）
這兩種情況也適用于集群環(huán)境。
嚴(yán)格針對(duì)您的情況，認(rèn)為以下步驟可能有助于解決您的問(wèn)題
為您的主題訂閱一個(gè)偵聽(tīng)器 listener-1。
過(guò)程-1
為消息 msg-1 添加狀態(tài)為“待發(fā)送”的數(shù)據(jù)庫(kù)條目
向主題發(fā)送消息 msg-1。在任何主題失敗的情況下重試發(fā)送如果在某些重試后步驟 2 失敗，process-1 必須在發(fā)送任何新消息之前重新發(fā)送 msg-1 或回滾步驟 1
聽(tīng)眾-1
使用訂閱的偵聽(tīng)器，從主題讀取參考（meesageID/correlationID），并將數(shù)據(jù)庫(kù)狀態(tài)更新為已發(fā)送，并從主題讀取/刪除消息。如果參考讀取成功并且數(shù)據(jù)庫(kù)更新失敗，主題仍然有消息。所以下一次讀取將更新數(shù)據(jù)庫(kù)。Incase 數(shù)據(jù)庫(kù)更新成功但消息刪除失敗。聽(tīng)眾將再次閱讀并嘗試更新已經(jīng)完成的消息。所以驗(yàn)證后可以忽略。
Incase listener 本身宕機(jī)，topic 將有消息，直到 listener 閱讀消息。在此之前，SENT 消息將處于“待發(fā)送”狀態(tài)。
慕俠2389804
TA貢獻(xiàn)1719條經(jīng)驗(yàn) 獲得超6個(gè)贊
在偵聽(tīng)器中保存數(shù)據(jù)庫(kù)行，其中包含字段 staus='pending'
另一個(gè)作業(yè)（獨(dú)立線程）將從數(shù)據(jù)庫(kù)中獲取所有待處理的行，并對(duì)每一行進(jìn)行以下操作：
2.1 將數(shù)據(jù)發(fā)送到主題
2.2 保存到數(shù)據(jù)庫(kù)中
如果我們?cè)诘?1 步失敗- 一切正常 - 數(shù)據(jù)處于一致?tīng)顟B(tài)，因?yàn)樽鳂I(yè)不會(huì)知道該數(shù)據(jù)的任何信息
如果我們?cè)诓襟E 2.1 上失敗了——沒(méi)問(wèn)題，下一個(gè)作業(yè)調(diào)用將嘗試處理它
如果我們?cè)诓襟E 2.2 上失敗了——如果我們?cè)谶@里失敗了——這意味著下一個(gè)作業(yè)調(diào)用將再次處理相同的數(shù)據(jù)。乍一看你可以認(rèn)為這是一個(gè)問(wèn)題。但是您的消費(fèi)者必須是冪等的——這意味著它必須了解消息已經(jīng)被處理并跳過(guò)處理。此要求是所有消息代理都保證消息將至少傳遞一次的結(jié)果。因此，無(wú)論如何，我們的消費(fèi)者都必須為重復(fù)的消息做好準(zhǔn)備。沒(méi)問(wèn)題了。
添加回答
舉報(bào)
0/150
提交
取消