ActiveMq筆記1-消息可靠性理論


原博客:http://shift-alt-ctrl.iteye.com/blog/2020182

      https://mp.weixin.qq.com/s/h74d6LtGB5M8VF0oLrXdCA

我們先看一消息的聲明周期如下圖:

圖片中簡單的描述了一條消息的生命周期,不過在不同的架構環境中,message的流動行可能更加復雜.將在稍后有關broker的架構中詳解..一條消息從producer端發出之后,一旦被broker正確保存,那么它將會被consumer消費,然后ACK,broker端才會刪除;不過當消息過期或者存儲設備溢出時,也會終結它。

1.ACK機制

 

  JMS API中約定了Client端可以使用四種ACK模式,在javax.jms.Session接口中:

 

  • AUTO_ACKNOWLEDGE = 1    自動確認 
  • CLIENT_ACKNOWLEDGE = 2    客戶端手動確認 如下圖:
  • 如果不寫這行代碼,會默認即消費失敗的,就不會出隊列。會一直存着,知道你ACK確認后。注意:JMS里面已經幫你封裝過了,無論你在配置中如何配置都不會生效的JMS源碼里面的配置如下圖:

  •  

  •  

  •  

  • DUPS_OK_ACKNOWLEDGE = 3    自動批量確認
  • SESSION_TRANSACTED = 0    事務提交並確認

    此外AcitveMQ補充了一個自定義的ACK模式:

  • INDIVIDUAL_ACKNOWLEDGE = 4    單條消息確認

  我們在開發JMS應用程序的時候,會經常使用到上述ACK模式,其中"INDIVIDUAL_ACKNOWLEDGE "只有ActiveMQ支持,當然開發者也可以使用它. ACK模式描述了Consumer與broker確認消息的方式(時機),比如當消息被Consumer接收之后,Consumer將在何時確認消息。對於broker而言,只有接收到ACK指令,才會認為消息被正確的接收或者處理成功了,通過ACK,可以在consumer(/producer)與Broker之間建立一種簡單的“擔保”機制. 

   

    Client端指定了ACK模式,但是在Client與broker在交換ACK指令的時候,還需要告知ACK_TYPE,ACK_TYPE表示此確認指令的類型,不同的ACK_TYPE將傳遞着消息的狀態,broker可以根據不同的ACK_TYPE對消息進行不同的操作。

 

    比如Consumer消費消息時出現異常,就需要向broker發送ACK指令,ACK_TYPE為"REDELIVERED_ACK_TYPE",那么broker就會重新發送此消息。在JMS API中並沒有定義ACT_TYPE,因為它通常是一種內部機制,並不會面向開發者。ActiveMQ中定義了如下幾種ACK_TYPE(參看MessageAck類):

 

  • DELIVERED_ACK_TYPE = 0    消息"已接收",但尚未處理結束
  • STANDARD_ACK_TYPE = 2    "標准"類型,通常表示為消息"處理成功",broker端可以刪除消息了
  • POSION_ACK_TYPE = 1    消息"錯誤",通常表示"拋棄"此消息,比如消息重發多次后,都無法正確處理時,消息將會被刪除或者DLQ(死信隊列)
  • REDELIVERED_ACK_TYPE = 3    消息需"重發",比如consumer處理消息時拋出了異常,broker稍后會重新發送此消息
  • INDIVIDUAL_ACK_TYPE = 4    表示只確認"單條消息",無論在任何ACK_MODE下    
  • UNMATCHED_ACK_TYPE = 5    在Topic中,如果一條消息在轉發給“訂閱者”時,發現此消息不符合Selector過濾條件,那么此消息將 不會轉發給訂閱者,消息將會被存儲引擎刪除(相當於在Broker上確認了消息)。

    到目前為止,我們已經清楚了大概的原理: Client端在不同的ACK模式時,將意味着在不同的時機發送ACK指令,每個ACK Command中會包含ACK_TYPE,那么broker端就可以根據ACK_TYPE來決定此消息的后續操作. 接下來,我們詳細的分析ACK模式與ACK_TYPE.

 

我們需要在創建Session時指定ACK模式,由此可見,ACK模式將是session共享的,意味着一個session下所有的 consumer都使用同一種ACK模式。在創建Session時,開發者不能指定除ACK模式列表之外的其他值.如果此session為事務類型,用戶指定的ACK模式將被忽略,而強制使用"SESSION_TRANSACTED"類型;如果session非事務類型時,也將不能將 ACK模式設定為"SESSION_TRANSACTED",畢竟這是相悖的.   

 Consumer消費消息的風格有2種: 同步/異步..使用consumer.receive()就是同步,使用messageListener就是異步;在同一個consumer中,我們不能同時使用這2種風格,比如在使用listener的情況下,當調用receive()方法將會獲得一個Exception。兩種風格下,消息確認時機有所不同。

 

同步調用時,在消息從receive方法返回之前,就已經調用了ACK;因此如果Client端沒有處理成功,此消息將丟失(可能重發,與ACK模式有關)。

基於異步調用時,消息的確認是在onMessage方法返回之后,如果onMessage方法異常,會導致消息不能被ACK,會觸發重發。

 

ACK模式詳解

 

 AUTO_ACKNOWLEDGE : 自動確認,這就意味着消息的確認時機將有consumer擇機確認."擇機確認"似乎充滿了不確定性,這也意味着,開發者必須明確知道"擇機確認"的具體時機,否則將有可能導致消息的丟失,或者消息的重復接收.那么在ActiveMQ中,AUTO_ACKNOWLEDGE是如何運作的呢?

    1) 對於consumer而言,optimizeAcknowledge屬性只會在AUTO_ACK模式下有效。

 

    2) 其中DUPS_ACKNOWLEGE也是一種潛在的AUTO_ACK,只是確認消息的條數和時間上有所不同。

 

    3) 在“同步”(receive)方法返回message之前,會檢測optimizeACK選項是否開啟,如果沒有開啟,此單條消息將立即確認,所以在這種情況下,message返回之后,如果開發者在處理message過程中出現異常,會導致此消息也不會redelivery,即"潛在的消息丟失";如果開啟了optimizeACK,則會在unAck數量達到prefetch * 0.65時確認,當然我們可以指定prefetchSize = 1來實現逐條消息確認。

 

    4) 在"異步"(messageListener)方式中,將會首先調用listener.onMessage(message),此后再ACK,如果onMessage方法異常,將導致client端補充發送一個ACK_TYPE為REDELIVERED_ACK_TYPE確認指令;如果onMessage方法正常,消息將會正常確認(STANDARD_ACK_TYPE)。此外需要注意,消息的重發次數是有限制的,每條消息中都會包含“redeliveryCounter”計數器,用來表示此消息已經被重發的次數,如果重發次數達到閥值,將會導致發送一個ACK_TYPE為POSION_ACK_TYPE確認指令,這就導致broker端認為此消息無法消費,此消息將會被刪除或者遷移到"dead letter"通道中。

    

    因此當我們使用messageListener方式消費消息時,通常建議在onMessage方法中使用try-catch,這樣可以在處理消息出錯時記錄一些信息,而不是讓consumer不斷去重發消息;如果你沒有使用try-catch,就有可能會因為異常而導致消息重復接收的問題,需要注意你的onMessage方法中邏輯是否能夠兼容對重復消息的判斷

 

CLIENT_ACKNOWLEDGE : 客戶端手動確認,這就意味着AcitveMQ將不會“自作主張”的為你ACK任何消息,開發者需要自己擇機確認。在此模式下,開發者需要需要關注幾個方法:1) message.acknowledge(),2) ActiveMQMessageConsumer.acknowledege(),3) ActiveMQSession.acknowledge();其1)和3)是等效的,將當前session中所有consumer中尚未ACK的消息都一起確認,2)只會對當前consumer中那些尚未確認的消息進行確認。開發者可以在合適的時機必須調用一次上述方法。為了避免混亂,對於這種ACK模式下,建議一個session下只有一個consumer。

 

    我們通常會在基於Group(消息分組)情況下會使用CLIENT_ACKNOWLEDGE,我們將在一個group的消息序列接受完畢之后確認消息(組);不過當你認為消息很重要,只有當消息被正確處理之后才能確認時,也可以使用此模式  。

 

    如果開發者忘記調用acknowledge方法,將會導致當consumer重啟后,會接受到重復消息,因為對於broker而言,那些尚未真正ACK的消息被視為“未消費”。

    開發者可以在當前消息處理成功之后,立即調用message.acknowledge()方法來"逐個"確認消息,這樣可以盡可能的減少因網絡故障而導致消息重發的個數;當然也可以處理多條消息之后,間歇性的調用acknowledge方法來一次確認多條消息,減少ack的次數來提升consumer的效率,不過這仍然是一個利弊權衡的問題。

 

    除了message.acknowledge()方法之外,ActiveMQMessageConumser.acknowledge()和ActiveMQSession.acknowledge()也可以確認消息,只不過前者只會確認當前consumer中的消息。其中sesson.acknowledge()和message.acknowledge()是等效的。

 

    無論是“同步”/“異步”,ActiveMQ都不會發送STANDARD_ACK_TYPE,直到message.acknowledge()調用。如果在client端未確認的消息個數達到prefetchSize * 0.5時,會補充發送一個ACK_TYPE為DELIVERED_ACK_TYPE的確認指令,這會觸發broker端可以繼續push消息到client端。(參看PrefetchSubscription.acknwoledge方法)

 

    在broker端,針對每個Consumer,都會保存一個因為"DELIVERED_ACK_TYPE"而“拖延”的消息個數,這個參數為prefetchExtension,事實上這個值不會大於prefetchSize * 0.5,因為Consumer端會嚴格控制DELIVERED_ACK_TYPE指令發送的時機(參見ActiveMQMessageConsumer.ackLater方法),broker端通過“prefetchExtension”與prefetchSize互相配合,來決定即將push給client端的消息個數,count = prefetchExtension + prefetchSize - dispatched.size(),其中dispatched表示已經發送給client端但是還沒有“STANDARD_ACK_TYPE”的消息總量;由此可見,在CLIENT_ACK模式下,足夠快速的調用acknowledge()方法是決定consumer端消費消息的速率;如果client端因為某種原因導致acknowledge方法未被執行,將導致大量消息不能被確認,broker端將不會push消息,事實上client端將處於“假死”狀態,而無法繼續消費消息。我們要求client端在消費1.5*prefetchSize個消息之前,必須acknowledge()一次;通常我們總是每消費一個消息調用一次,這是一種良好的設計。

 

    此外需要額外的補充一下:所有ACK指令都是依次發送給broker端,在CLIET_ACK模式下,消息在交付給listener之前,都會首先創建一個DELIVERED_ACK_TYPE的ACK指令,直到client端未確認的消息達到"prefetchSize * 0.5"時才會發送此ACK指令,如果在此之前,開發者調用了acknowledge()方法,會導致消息直接被確認(STANDARD_ACK_TYPE)。broker端通常會認為“DELIVERED_ACK_TYPE”確認指令是一種“slow consumer”信號,如果consumer不能及時的對消息進行acknowledge而導致broker端阻塞,那么此consumer將會被標記為“slow”,此后queue中的消息將會轉發給其他Consumer。

 

    DUPS_OK_ACKNOWLEDGE : "消息可重復"確認,意思是此模式下,可能會出現重復消息,並不是一條消息需要發送多次ACK才行。它是一種潛在的"AUTO_ACK"確認機制,為批量確認而生,而且具有“延遲”確認的特點。對於開發者而言,這種模式下的代碼結構和AUTO_ACKNOWLEDGE一樣,不需要像CLIENT_ACKNOWLEDGE那樣調用acknowledge()方法來確認消息。

 

    1) 在ActiveMQ中,如果在Destination是Queue通道,我們真的可以認為DUPS_OK_ACK就是“AUTO_ACK + optimizeACK + (prefetch > 0)”這種情況,在確認時機上幾乎完全一致;此外在此模式下,如果prefetchSize =1 或者沒有開啟optimizeACK,也會導致消息逐條確認,從而失去批量確認的特性。

 

    2) 如果Destination為Topic,DUPS_OK_ACKNOWLEDGE才會產生JMS規范中詮釋的意義,即無論optimizeACK是否開啟,都會在消費的消息個數>=prefetch * 0.5時,批量確認(STANDARD_ACK_TYPE),在此過程中,不會發送DELIVERED_ACK_TYPE的確認指令,這是1)和AUTO_ACK的最大的區別。

 

    這也意味着,當consumer故障重啟后,那些尚未ACK的消息會重新發送過來。

 

    SESSION_TRANSACTED : 當session使用事務時,就是使用此模式。在事務開啟之后,和session.commit()之前,所有消費的消息,要么全部正常確認,要么全部redelivery。這種嚴謹性,通常在基於GROUP(消息分組)或者其他場景下特別適合。在SESSION_TRANSACTED模式下,optimizeACK並不能發揮任何效果,因為在此模式下,optimizeACK會被強制設定為false,不過prefetch仍然可以決定DELIVERED_ACK_TYPE的發送時機。

 

    因為Session非線程安全,那么當前session下所有的consumer都會共享同一個transactionContext;同時建議,一個事務類型的Session中只有一個Consumer,以避免rollback()或者commit()方法被多個consumer調用而造成的消息混亂。

    

    當consumer接受到消息之后,首先檢測TransactionContext是否已經開啟,如果沒有,就會開啟並生成新的transactionId,並把信息發送給broker;此后將檢測事務中已經消費的消息個數是否 >= prefetch * 0.5,如果大於則補充發送一個“DELIVERED_ACK_TYPE”的確認指令;這時就開始調用onMessage()方法,如果是同步(receive),那么即返回message。上述過程,和其他確認模式沒有任何特殊的地方。

   

    當開發者決定事務可以提交時,必須調用session.commit()方法,commit方法將會導致當前session的事務中所有消息立即被確認;事務的確認過程中,首先把本地的deliveredMessage隊列中尚未確認的消息全部確認(STANDARD_ACK_TYPE);此后向broker發送transaction提交指令並等待broker反饋,如果broker端事務操作成功,那么將會把本地deliveredMessage隊列清空,新的事務開始;如果broker端事務操作失敗(此時broker已經rollback),那么對於session而言,將執行inner-rollback,這個rollback所做的事情,就是將當前事務中的消息清空並要求broker重發(REDELIVERED_ACK_TYPE),同時commit方法將拋出異常。

 

    當session.commit方法異常時,對於開發者而言通常是調用session.rollback()回滾事務(事實上開發者不調用也沒有問題),當然你可以在事務開始之后的任何時機調用rollback(),rollback意味着當前事務的結束,事務中所有的消息都將被重發。需要注意,無論是inner-rollback還是調用session.rollback()而導致消息重發,都會導致message.redeliveryCounter計數器增加,最終都會受限於brokerUrl中配置的"jms.redeliveryPolicy.maximumRedeliveries",如果rollback的次數過多,而達到重發次數的上限時,消息將會被DLQ(dead letter)。

 

    INDIVIDUAL_ACKNOWLEDGE : 單條消息確認,這種確認模式,我們很少使用,它的確認時機和CLIENT_ACKNOWLEDGE幾乎一樣,當消息消費成功之后,需要調用message.acknowledege來確認此消息(單條),而CLIENT_ACKNOWLEDGE模式先message.acknowledge()方法將導致整個session中所有消息被確認(批量確認)。

 

 optimizeACK

  "可優化的ACK",這是ActiveMQ對於consumer在消息消費時,對消息ACK的優化選項,也是consumer端最重要的優化參數之一,

1) 在brokerUrl中增加如下查詢字符串: 

  

  

String brokerUrl = "tcp://localhost:61616?" +   "jms.optimizeAcknowledge=true" +   "&jms.optimizeAcknowledgeTimeOut=30000" +   "&jms.redeliveryPolicy.maximumRedeliveries=6";  

  ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerUrl); 

 2) 在destinationUri中,增加如下查詢字符串:

  

String queueName = "test-queue?customer.prefetchSize=100";  
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
Destination queue = session.createQueue(queueName); 

我們需要在brokerUrl指定optimizeACK選項,在destinationUri中指定prefetchSize(預獲取)選項,其中brokerUrl參數選項是全局的,即當前factory下所有的connection/session/consumer都會默認使用這些值;而destinationUri中的選項,只會在使用此destination的consumer實例中有效;如果同時指定,brokerUrl中的參數選項值將會被覆蓋。optimizeAck表示是否開啟“優化ACK”,只有在為true的情況下,prefetchSize(下文中將會簡寫成prefetch)以及optimizeAcknowledgeTimeout參數才會有意義。此處需要注意"optimizeAcknowledgeTimeout"選項只能在brokerUrl中配置。

    prefetch值建議在destinationUri中指定,因為在brokerUrl中指定比較繁瑣;在brokerUrl中,queuePrefetchSize和topicPrefetchSize都需要單獨設定:"&jms.prefetchPolicy.queuePrefetch=12&jms.prefetchPolicy.topicPrefetch=12"等來逐個指定。

 

    如果prefetchACK為true,那么prefetch必須大於0;當prefetchACK為false時,你可以指定prefetch為0以及任意大小的正數。不過,當prefetch=0是,表示consumer將使用PULL(拉取)的方式從broker端獲取消息,broker端將不會主動push消息給client端,直到client端發送PullCommand時;當prefetch>0時,就開啟了broker push模式,此后只要當client端消費且ACK了一定的消息之后,會立即push給client端多條消息。

 

    當consumer端使用receive()方法同步獲取消息時,prefetch可以為0和任意正值;當prefetch=0時,那么receive()方法將會首先發送一個PULL指令並阻塞,直到broker端返回消息為止,這也意味着消息只能逐個獲取(類似於Request<->Response),這也是Activemq中PULL消息模式;當prefetch > 0時,broker端將會批量push給client 一定數量的消息(<= prefetch),client端會把這些消息(unconsumedMessage)放入到本地的隊列中,只要此隊列有消息,那么receive方法將會立即返回,當一定量的消息ACK之后,broker端會繼續批量push消息給client端。

 

    當consumer端使用MessageListener異步獲取消息時,這就需要開發設定的prefetch值必須 >=1,即至少為1;在異步消費消息模式中,設定prefetch=0,是相悖的,也將獲得一個Exception。

 

    此外,我們還可以brokerUrl中配置“redelivery”策略,比如當一條消息處理異常時,broker端可以重發的最大次數;和下文中提到REDELIVERED_ACK_TYPE互相協同。當消息需要broker端重發時,consumer會首先在本地的“deliveredMessage隊列”(Consumer已經接收但還未確認的消息隊列)刪除它,然后向broker發送“REDELIVERED_ACK_TYPE”類型的確認指令,broker將會把指令中指定的消息重新添加到pendingQueue(亟待發送給consumer的消息隊列)中,直到合適的時機,再次push給client。

 

    到目前為止,或許你知道了optimizeACK和prefeth的大概意義,不過我們可能還會有些疑惑!!optimizeACK和prefetch配合,將會達成一個高效的消息消費模型:批量獲取消息,並“延遲”確認(ACK)prefetch表達了“批量獲取”消息的語義,broker端主動的批量push多條消息給client端,總比client多次發送PULL指令然后broker返回一條消息的方式要優秀很多,它不僅減少了client端在獲取消息時阻塞的次數和阻塞的時間,還能夠大大的減少網絡開支。optimizeACK表達了“延遲確認”的語義(ACK時機),client端在消費消息后暫且不發送ACK,而是把它緩存下來(pendingACK),等到這些消息的條數達到一定閥值時,只需要通過一個ACK指令把它們全部確認;這比對每條消息都逐個確認,在性能上要提高很多。由此可見,prefetch優化了消息傳送的性能,optimizeACK優化了消息確認的性能。

 

    當consumer端消息消費的速率很高(相對於producer生產消息),而且消息的數量也很大時(比如消息源源不斷的生產),我們使用optimizeACK + prefetch將會極大的提升consumer的性能。不過反過來:

    1) 如果consumer端消費速度很慢(對消息的處理是耗時的),過大的prefetchSize,並不能有效的提升性能,反而不利於consumer端的負載均衡(只針對queue);按照良好的設計准則,當consumer消費速度很慢時,我們通常會部署多個consumer客戶端,並使用較小的prefetch,同時關閉optimizeACK,可以讓消息在多個consumer間“負載均衡”(即均勻的發送給每個consumer);如果較大的prefetchSize,將會導致broker一次性push給client大量的消息,但是這些消息需要很久才能ACK(消息積壓),而且在client故障時,還會導致這些消息的重發。

 

    2) 如果consumer端消費速度很快,但是producer端生成消息的速率較慢,比如生產者10秒鍾生成10條消息,但是consumer一秒就能消費完畢,而且我們還部署了多個consumer!!這種場景下,建議開啟optimizeACK,但是需要設置的prefetchSize不能過大;這樣可以保證每個consumer都能有"活干",否則將會出現一個consumer非常忙碌,但是其他consumer幾乎收不到消息。

 

    3) 如果消息很重要,特別是不願意接收到”redelivery“的消息,那么我們需要將optimizeACK=false,prefetchSize=1

 

    既然optimizeACK是”延遲“確認,那么就引入一種潛在的風險:在消息被消費之后還沒有來得及確認時,client端發生故障,那么這些消息就有可能會被重新發送給其他consumer,那么這種風險就需要client端能夠容忍“重復”消息。

 

    prefetch值默認為1000,當然這個值可能在很多場景下是偏大的;我們暫且不考慮ACK模式(參見下文),通常情況下,我們只需要簡單的統計出單個consumer每秒的最大消費消息數即可,比如一個consumer每秒可以處理100個消息,我們期望consumer端每2秒確認一次,那么我們的prefetchSize可以設置為100 * 2 /0.65大概為300。無論如何設定此值,client持有的消息條數最大為:prefetch + “DELIVERED_ACK_TYPE消息條數”(DELIVERED_ACK_TYPE參見下文)

 

     即使當optimizeACK為true,也只會當session的ACK模式為AUTO_ACKNOWLEDGE時才會生效,即在其他類型的ACK模式時consumer端仍然不會“延遲確認”,即:

  

consumer.optimizeAck = connection.optimizeACK && session.isAutoAcknowledge()

 當consumer.optimizeACK有效時,如果客戶端已經消費但尚未確認的消息(deliveredMessage)達到prefetch * 0.65,consumer端將會自動進行ACK;同時如果離上一次ACK的時間間隔,已經超過"optimizeAcknowledgeTimout"毫秒,也會導致自動進行ACK。

 

    此外簡單的補充一下,批量確認消息時,只需要在ACK指令中指明“firstMessageId”和“lastMessageId”即可,即消息區間,那么broker端就知道此consumer(根據consumerId識別)需要確認哪些消息。

 

  最后讓我們想一下,如何保證消息隊列做最終一致性的時候消息不會丟失和重復消費呢?

    我們可以通過冪等的方式實現。所謂的冪等就是多次調用,結果和一次調用一致。即我們通過生產者發送消息帶一個MessageID來到Broker,Broker已經收到了,返回ACK給生產者,如果此時發生網絡故障了,生產者沒收到ACK,生產者以為Broker沒收到ACK,生產者重傳消息。這樣造成重復消費,如果我們通過通過生產者發送消息帶一個MessageID來到Broker,Broker記錄這個MessageID,Broker已經收到了,返回ACK給生產者,如果此時發生網絡故障了,生產者沒收到ACK,生產者以為Broker沒收到ACK,生產者重傳消息並再次附帶MessageID,Broker根據這個MessageID核對本地是否有這個MessageID,如果有,返回ACK給生產者,對生產者再次發來的Message丟棄不管。這樣就解決了隊列做最終一致性的時候消息不會丟失和重復消費。

  一、緣起

  《消息總線消息必達》所述,MQ消息必達,架構上有兩個核心設計點:

  (1)消息落地

  (2)消息超時、重傳、確認

 

  

 

  再次回顧消息總線核心架構,它由發送端、服務端、固化存儲、接收端四大部分組成。

 

  為保證消息的可達性,超時、重傳、確認機制可能導致消息總線、或者業務方收到重復的消息,從而對業務產生影響。

 

  舉個例子:

  購買會員卡,上游支付系統負責給用戶扣款,下游系統負責給用戶發卡,通過MQ異步通知。不管是上半場的ACK丟失,導致MQ收到重復的消息,還是下半場ACK丟失,導致購卡系統收到重復的購卡通知,都可能出現,上游扣了一次錢,下游發了多張卡。

 

  消息總線的冪等性設計至關重要,是本文將要討論的重點。

 

  二、上半場的冪等性設計

  

 

  MQ消息發送上半場,即上圖中的1-3

  1,發送端MQ-client將消息發給服務端MQ-server

  2,服務端MQ-server將消息落地

  3,服務端MQ-server回ACK給發送端MQ-client

  如果3丟失,發送端MQ-client超時后會重發消息,可能導致服務端MQ-server收到重復消息。

 

  此時重發是MQ-client發起的,消息的處理是MQ-server,為了避免步驟2落地重復的消息,對每條消息,MQ系統內部必須生成一個inner-msg-id,作為去重和冪等的依據,這個內部消息ID的特性是:

  (1)全局唯一

  (2)MQ生成,具備業務無關性,對消息發送方和消息接收方屏蔽

 

  有了這個inner-msg-id,就能保證上半場重發,也只有1條消息落到MQ-server的DB中,實現上半場冪等。

 

三、下半場的冪等性設計

 

MQ消息發送下半場,即上圖中的4-6

4,服務端MQ-server將消息發給接收端MQ-client

5,接收端MQ-client回ACK給服務端

6,服務端MQ-server將落地消息刪除

需要強調的是,接收端MQ-client回ACK給服務端MQ-server,是消息消費業務方的主動調用行為,不能由MQ-client自動發起,因為MQ系統不知道消費方什么時候真正消費成功。

如果5丟失,服務端MQ-server超時后會重發消息,可能導致MQ-client收到重復的消息。

 

此時重發是MQ-server發起的,消息的處理是消息消費業務方,消息重發勢必導致業務方重復消費(上例中的一次付款,重復發卡),為了保證業務冪等性,業務消息體中,必須有一個biz-id,作為去重和冪等的依據,這個業務ID的特性是:

(1)對於同一個業務場景,全局唯一

(2)由業務消息發送方生成,業務相關,對MQ透明

(3)由業務消息消費方負責判重,以保證冪等

 

最常見的業務ID有:支付ID,訂單ID,帖子ID等。

 

具體到支付購卡場景,發送方必須將支付ID放到消息體中,消費方必須對同一個支付ID進行判重,保證購卡的冪等。

 

有了這個業務ID,才能夠保證下半場消息消費業務方即使收到重復消息,也只有1條消息被消費,保證了冪等。

 

三、總結

MQ為了保證消息必達,消息上下半場均可能發送重復消息,如何保證消息的冪等性呢?

上半場

MQ-client生成inner-msg-id,保證上半場冪等。

這個ID全局唯一,業務無關,由MQ保證。

 

下半場

業務發送方帶入biz-id,業務接收方去重保證冪等。

這個ID對單業務唯一,業務相關,對MQ透明。

 

結論:冪等性,不僅對MQ有要求,對業務上下游也有要求。

  

 

 

    結語:到目前為止,我們已經已經簡單的了解了ActiveMQ中消息傳送機制,還有JMS中ACK策略,重點分析了optimizeACK的策略,希望開發者能夠在使用activeMQ中避免一些不必要的錯誤。

 


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM