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。两种风格下,消息确认时机有所不同。

"同步"伪代码:

Java代码  

  1. //receive伪代码---过程  
  2. Message message = sessionMessageQueue.dequeue();  
  3. if(message != null){  
  4.     ack(message);  
  5. }  
  6. return message  

      同步调用时,在消息从receive方法返回之前,就已经调用了ACK;因此如果Client端没有处理成功,此消息将丢失(可能重发,与ACK模式有关)。

"异步"伪代码:

Java代码  

  1. //基于listener  
  2. Session session = connection.getSession(consumerId);  
  3. sessionQueueBuffer.enqueue(message);  
  4. Runnable runnable = new Ruannale(){  
  5.     run(){  
  6.         Consumer consumer = session.getConsumer(consumerId);  
  7.         Message md = sessionQueueBuffer.dequeue();  
  8. try{  
  9.             consumer.messageListener.onMessage(md);  
  10.             ack(md);//  
  11.         }catch(Exception e){  
  12.             redelivery();//sometime,not all the time;  
  13.     }  
  14. }  
  15. //session中将采取线程池的方式,分发异步消息  
  16. //因此同一个session中多个consumer可以并行消费  
  17. 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

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Spark学习技巧

高性能:MYSQL异步客户端

实时处理领域,当需要使用外部存储数据染色的时候,需要慎重对待,不能让与外部系统之间的交互延迟对流的整个进度取决定性的影响。

3902
来自专栏何俊林

一个强大的网络库RxEasyHttp

本文来自周游的投稿,封装的网络请求库,github:https://github.com/zhou-you/RxEasyHttp,欢迎点击【阅读原文】,查看更多...

3918
来自专栏散尽浮华

进程管理利器-supervisor部署记录

一、简单介绍 supervisor是用来管理进程的一个工具,止于为什么要用supervisor,是因为相对于linux传统的进程管理方式来说,它有很多的优势: ...

3838
来自专栏微服务

SQL Server 存储过程

Transact-SQL中的存储过程,非常类似于Java语言中的方法,它可以重复调用。当存储过程执行一次后,可以将语句缓存中,这样下次执行的时候直接使用缓存中的...

3519
来自专栏我是攻城师

关于Hbase多版本存储的一个注意点

47416
来自专栏hbbliyong

SQL Server 存储过程

Transact-SQL中的存储过程,非常类似于Java语言中的方法,它可以重复调用。当存储过程执行一次后,可以将语句缓存中,这样下次执行的时候直接使用缓存中的...

4048
来自专栏实用工具入门教程

如何部署 MySql 集群

MySQL Cluster分布式数据库为MySQL数据库管理系统提供高可用性和吞吐量。MySQL Cluster由一个或多个管理节点(ndb_mgmd)组成,这...

3075
来自专栏linux、Python学习

Linux基础教程之linux文件权限深度解读

基本命令—— 1.cut : cat /etc/passwd | cut -d’:’ -f7| uniq -c| sort -nr

770
来自专栏自由而无用的灵魂的碎碎念

关于Windows Vista下几大还原技术比较

注:本文中讨论到的一部分技术可能只有在Windows Vista  Business/Ultimate下才包含。本文直接以Ultimate版本讨论。

1366
来自专栏我是攻城师

elasticsearch里面的关于批量读取mget的用法

4069

扫码关注云+社区

领取腾讯云代金券