JAVA消息確認機制之ACK模式

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

• AUTO_ACKNOWLEDGE = 1    自動確認
• CLIENT_ACKNOWLEDGE = 2    客戶端手動確認   
• 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.

Java代碼  

　　Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);   

       我們需要在創建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偽代碼---過程
Message message = sessionMessageQueue.dequeue();
if(message != null){
    ack(message);
}
return message

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

 

      "異步"偽代碼：

//基於listener
Session session = connection.getSession(consumerId);
sessionQueueBuffer.enqueue(message);
Runnable runnable = new Ruannale(){
    run(){
        Consumer consumer = session.getConsumer(consumerId);
        Message md = sessionQueueBuffer.dequeue();
        try{
            consumer.messageListener.onMessage(md);
            ack(md);//
        }catch(Exception e){
            redelivery();//sometime，not all the time;
    }
}
//session中將采取線程池的方式，分發異步消息
//因此同一個session中多個consumer可以並行消費
threadPool.execute(runnable);

      基於異步調用時，消息的確認是在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中所有消息被確認(批量確認)。

    結語：到目前為止，我們已經已經簡單的了解了ActiveMQ中消息傳送機制，還有JMS中ACK策略，重點分析了optimizeACK的策略，希望開發者能夠在使用activeMQ中避免一些不必要的錯誤。本文如有疏漏和錯誤之處，請各位不吝賜教，特此感謝。

 

 

參考:
http://shift-alt-ctrl.iteye.com/blog/2020182
http://shift-alt-ctrl.iteye.com/blog/2020182

 我的博客即將搬運同步至騰訊雲+社區，邀請大家一同入駐：https://cloud.tencent.com/developer/support-plan?invite_code=2xsv7f2sjiqs8