active mq 事物以及ack模式事务介绍

柔情只为你懂 2022-05-16 05:42 94阅读 0赞

active mq 事物以及ack模式 事务介绍

ACK模式与类型介绍

JMS API中约定了Client端可以使用四种ACK\_MODE,在javax.jms.Session接口中:

AUTO\_ACKNOWLEDGE= 1          自动确认   
CLIENT\_ACKNOWLEDGE= 2       客户端手动确认   
DUPS\_OK\_ACKNOWLEDGE= 3   自动批量确认   
SESSION\_TRANSACTED= 0        事务提交并确认

![70][]

在与Spring MVC集成中 mq源码默认 使用AUTO\_ACKNOWLEDGE

![70 1][]

![70 2][]

![70 3][]

Session.AUTO\_ACKNOWLEDGE为自动确认,客户端发送和接收消息不需要做额外的工作。哪怕是接收端发生异常,也会被当作正常发送成功。 Session.CLIENT\_ACKNOWLEDGE为客户端确认。客户端接收到消息后,必须调用javax.jms.Message的acknowledge方法。jms服务器才会当作发送成功,并删除消息。 DUPS\_OK\_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回,会话对象就会确认消息的接收;而且允许重复确认。

![70 4][]

mq消息手动确认设置如上。

此外AcitveMQ补充了一个自定义的ACK\_MODE:

INDIVIDUAL\_ACKNOWLEDGE = 4    单条消息确认

我们在开发JMS应用程序的时候,会经常使用到上述ACK\_MODE,其中"INDIVIDUAL\_ACKNOWLEDGE "只有ActiveMQ支持,当然开发者也可以使用它. ACK\_MODE描述了Consumer与broker确认消息的方式(时机),比如当消息被Consumer接收之后,Consumer将在何时确认消息。对于broker而言，只有接收到ACK指令,才会认为消息被正确的接收或者处理成功了,通过ACK，可以在consumer与Broker之间建立一种简单的“担保”机制.

Client端指定了ACK\_MODE,但是在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    BROKER间转发消息时,接收端"拒绝"消息

如图：

![70 5][]

到目前为止,我们已经清楚了大概的原理: Client端在不同的ACK\_MODE时,将意味着在不同的时机发送ACK指令,每个ACK Command中会包含ACK\_TYPE,那么broker端就可以根据ACK\_TYPE来决定此消息的后续操作. 接下来,我们详细的分析ACK\_MODE与ACK\_TYPE.

![70 6][]

我们需要在创建Session时指定ACK\_MODE,由此可见,ACK\_MODE将是session共享的,意味着一个session下所有的 consumer都使用同一种ACK\_MODE。在创建Session时,开发者不能指定除ACK\_MODE列表之外的其他值.如果此session为事务类型,用户指定的ACK\_MODE将被忽略,而强制使用"SESSION\_TRANSACTED"类型;如果session非事务类型时,也将不能将 ACK\_MODE设定为"SESSION\_TRANSACTED",毕竟这是相悖的.

如图：

![70 7][]

Consumer消费消息的风格有2种: 同步/异步..使用consumer.receive()就是同步，使用messageListener就是异步；在同一个consumer中，我们不能使用使用这2种风格，比如在使用listener的情况下，当调用receive()方法将会获得一个Exception。两种风格下，消息确认时机有所不同。

![70 8][]

同步调用时，

当在事务中，则需要手动commit  相当于调用了ACK；如上图：

当非事务中，则看你session 创建时的设置,比如：

//创建事物会话(事务支持事务,ack模式)
    Session session = connection.createSession(Boolean.FALSE, Session.CLIENT_ACKNOWLEDGE);

**   异步代码：**

//基于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\_MODE详解**

**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中那些尚未确认的消息进行确认（可以批量确认已经消费的哦）。开发者可以在合适的时机必须调用一次上述方法。

我们通常会在基于Group(消息分组)情况下会使用CLIENT\_ACKNOWLEDGE，我们将在一个group的消息序列接受完毕之后确认消息(组)；不过当你认为消息很重要，只有当消息被正确处理之后才能确认时，也很可以使用此ACK\_MODE。

如果开发者忘记调用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中避免一些不必要的错误。本文如有疏漏和错误之处，请各位不吝赐教，特此感谢。

源码参考类：

1) ActiveMQConnectionFactory,ActiveMQMessageConsumer,ActiveMQSession,MessageAck等

2) Queue,PrefetchSubscription,TransactionContext,TransactionStore

[70]: /images/20220516/f5540c0238b444cf81c8a948f3fd4970.png
[70 1]: /images/20220516/c9362188907041a3a33d7691d1204fbd.png
[70 2]: /images/20220516/6bca9485ed8542d0ba9ed649a076b265.png
[70 3]: /images/20220516/890ca809fc214d0b9ea8814872607624.png
[70 4]: /images/20220516/edbab1f540e546bf9865fc1af4fd8536.png
[70 5]: /images/20220516/6a46daa2b0054879a12589da92fbf8a1.png
[70 6]: /images/20220516/1071d7c1f1c94167bd4a241ee70f2b40.png
[70 7]: /images/20220516/33e4404cc3984ea68b6d1fef41d240f6.png
[70 8]: /images/20220516/9f2c34b4a8cc491aab00e087a62342b3.png