訊息中介軟體activeMQ（6）
訊息總是從生產者傳送到中介軟體再有中介軟體傳送給消費者。
訊息從broker傳送到消費者之後，為了使訊息能夠被正確的消費，引入了ACK機制來進行訊息的確認。
概念一：
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"等來逐個指定。
如果optimizeAck為true，那麼prefetch必須大於0；因為optimizeAck的做大作用是延時確認，如果pretchSize設定為0，即採用了pull的形式，完全達不到延遲確認的效果。
當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訊息條數”
即使當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識別)需要確認哪些訊息。
概念二. 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:
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 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端。 在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)”這種情況，在確認時機上幾乎完全一致（程式碼測試發現目的地是佇列的時候，無論怎樣設定optimizeACK 和prefetch 的值，訊息都是被逐條進行確認的批量確認的特性。）；此外在此模式下，如果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中所有訊息被確認(批量確認)。