文档章节

ActiveMQ中Consumer特征详解与优化《《转载

泡海椒
 泡海椒
发布于 2016/02/05 16:03
字数 7621
阅读 83
收藏 0

ActiveMQ中Consumer特征详解与优化

www.MyException.Cn  网友分享于:2015-08-26  浏览:129次

ActiveMQ中Consumer特性详解与优化
   

前言

    从本文中你可以了解到如下内容:

    1) consumer端消息消费的模型,session的运作机制

    2) 如果提升broker和consumer端消息消费的速率

    3) selector,group,exclusive对消息消费的影响

    4) 如何让Priority更好的运行,提高消息的顺序性

    5) Slow Consumer的产生原因,以及如何调优。

 

 

    Consumer作为ActiveMQ的消费端,开起来简单,不过还有很多隐藏的特性,值得我们去探索和调优。

    如下为典型的Consumer端代码示例

String brokerUrl = "tcp://localhost:61616?"
String queueName = "test-queue";
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerUrl);
ActiveMQConnection connection = (ActiveMQConnection)factory.createConnection();
final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination queue = session.createQueue(queueName);
MessageConsumer consumer = session.createConsumer(queue);
consumer.setMessageListener(new MessageListener() {
	@Override
	public void onMessage(Message message) {
		try{
			ActiveMQMessage m = (ActiveMQMessage)message;
			System.out.println(m.toString());
		}catch(Exception e){
			//
			e.printStackTrace();
		}
	}
});
connection.start();

//connection.close();

 

    Consumer端涉及到众多的特性,其中包括:消息转发机制,session机制与多线程,消息ACK策略等,我们无法详解所有的原理,如下为Consumer端消息转发机制,仅供参考,[详细请参考]:



 

    上图非常复杂,具体的原理,我们将会在下文中逐个解释。大概原理与过程:

 

  • 通过Connection实例创建Session之后,将会把session实例保存在本地的list中,即connection持有session列表,并在底层开启transport,侦听数据。

  • Session创建Consumer之后,将会把Consumer实例添加到本地的list中,以便此后分拣消息,即session持有 consumer列表;此外connection中也持有一个consumer集合(Map),其中Key为consumerId,value为 session引用。

  • 如果Session支持异步转发(asyncDispatch)或者使用了转发池(dispatchPool),将创建线程池用来转发消息。

  • 当broker端有消息通过transport发送时,connection将会分拣消息,根据消息中指定的consumerId,从本地session列表中获取其对应的session实例;然后将消息交付session负责转发。

  • 当session接受到消息后,会根据消息的consumerId,在本地consumer列表中找到对应的consumer实例;检测 session的消息转发方式,如果是同步转发,则直接将消息交给consumer,即调用consumer.dispatch(message),此方 法要么调用messageListener.onMessage(),要么将消息添加到本地的unconsumedMessages队列中(唤醒 receive);如果是异步转发,则将消息添加到session级别的队列中并由线程池负责转发。

  • 当consumer接受到消息之后,将会调用messageListener.onMessage方法或者从receive方法中返回。

  • 当消息消费成功后,consumer将会根据指定的ACK_MODE负责向broker发送ACK指令,此后消息将会在broker端清除。

 

1. asyncDispatch

    broker端是否允许使用“异步转发”。broker端处理connection、session、consumer的模型和Client端几乎一样; 当通道中(Queue/Topic)有消息需要转发给consumer时,将会调用相应的Subscription的dispatch方法(在 broker端负责与Consumer通讯的服务,称为Subscription,每个Consumer客户端对应broker端一个 Subscription实例);如果asyncDispatch为false,那么将会阻塞转发线程(dispatchThread),直到底层的 Transport将消息发送到Consumer客户端(如果底层transport正在发送消息,意味着阻塞);如果为true,则会将消息添加到 transport本地的队列中(负责与Consumer客户端通信的每个transport都有一个本地的queue和线程池),并启动线程池负责发送 消息,那么转发线程即可立即返回。

    (参见代码:QueueStoreCursor/TopicStoreCursor --> broker端Queue/Topic(核心 类)--->QueuePrefetchSubscription,QueueRegion,AbstractRegion,RoundRobinDispatchPolicy/StrictOrderDispatchPolicy)

    

    由此可见,当前通道中所有的consumer都在brorker端开启了“异步转发”,可以大大提升broker转发消息的性能,此值默认为true。

 

//在brokerUrl中设置
tcp://localhost:61616?jms.dispatchAsync=true
//在destinationUri中为当前通道设定
orderQueue?consumer.dispatchAsync=true

 

     broker“异步转发”采用的是线程池模式,线程池的缺点即使当任务单元运行时间足够短时,线程切换的成本将变得很重要,这也意味着,如果通道中 (Queue/Topic)中的consumer客户端接收和消费消息的速度很快,我们可以关闭异步转发来避免线程切换。如果consumer通常为慢速 (Slower Consumer),且关闭了异步转发,那就意味着每次主线程转发消息时,都需要阻塞较长的时间,这极大的影响了整体的通讯效率。无论如何,我们都没有理 由关闭异步转发,异步转发应该是最佳选择。

 

    备注:broker端Queue/Topic(例如:org.apache.activemq.broker.region.Queue)实例,负责从底 层store中获取消息,并使用线程池模式转发消息,dispatchPolicy负责将消息转发给相应的subscription,当Consumer 客户端创建时,会向Broker发送一个指令,这个指令中包括consumerId(参看ConsumerInfo),那么此后Region(在逻辑上相 当于客户端的Session,但每个通道只有一个实例)将会为每个ConsumerId创建相应Subscription实例(持有底层 Transport),那么此后Consumer与Broker的数据交互,就有Subscription负责。每个Region都维护了一个 Subscription列表,用于dispatchPolicy转发消息。这个设计模型,和Consumer客户端 + Session非常相似。

 

2. alwaysSessionAsync

    客户端session是否使用异步转发,它和(1)中的参数没有任何关系,当底层Transport接收到Broker发送的消息后,会交付给session,那么session是否采用异步的方式将消息传递给Consumer!此值默认为true。

    如果为false,表示使用同步。当consumer使用receive()获取消息时,那么session将会把消息添加到consumer的本地 queue,然后唤醒receive等待;当consumer使用messageListener异步侦听消息时,将会调用其onMessage()方法 直到方法执行完毕,然后返回。

    如果为true,表示使用异步。session接受的消息,将会首先放入session buffer中(队列),那么此后的线程池将会负责移除buffer中的消息,并转发给相应的consumer。

 

    当session中有多个consumer时,或者Transport中消息量比较密集,异步方式是最佳的。如果session中只有一个consumer,或者transport中消息量很少,使用异步并不能明显的提升性能。

 

//在brokerUrl中设置
tcp://localhost:61616?jms.alwaysSessionAsync=true

 

3. maxThreadPoolSize

    在alwaysSessionAsync为true的情况下,会引入一个线程池,那么这个线程池中线程的个数,默认为 Integer.MAX_VALUE,不过我们首先需要清楚,这个线程池是Connection内所有的session共享的。你可以指定合适的值来决定 线程池的大小。

 

//在brokerUrl中设置
tcp://localhost:61616?jms.maxThreadPoolSize=64

 

4. useDedicatedTaskRunner

    当前connetion下的每个session是否使用单独的线程。在上述介绍中,我知道了异步转发是使用了线程池的,而且这个线程池是当前 Connection下所有的session共享,尽管这不会带来任何问题,可是有时候我们可能期望在threadLocal或者基于当前线程绑定一些特 殊的数据,那么我们就可以使用useDedicatedTaskRunner=true;这就意味着,每个session都将创建一个单独的线程,整个生 命周期中都使用它来转发消息。

 

    当每个session只有一个Consumer,或者有多个consumer但它们的处理业务类似(比如selector具有一定逻辑关系),或者这些 consumer消费速度都很快,或者我们期望在Consumer中使用类似于Thread.sleep(long)这样的方式来阻塞session转发 消息时,这个参数可以帮助我们。

 

//brokerUrl中,默认为false
tcp://localhost:61616?jms.useDedicatedTaskRunner=true

  

5. redeliveryPolicy

    consumer使用的重发策略,当消息在client端处理失败(比如onMessage方法抛出异常,事务回滚等),将会触发消息重发。对于 Broker端,需要重发的消息将会被立即发送(如果broker端使用异步发送,且发送队列中还有其他消息,那么重发的消息可能不会被立即到达 Consumer)。我们通过此Policy配置最大重发次数、重发频率等,如果你的Consumer客户端处于不良网络环境中,可以适当调整相关参数。 参数列表,请参见(RedeliveryPolicy)

 

//在brokerUrl中设置
tcp://localhost:61616?jms.redeliveryPolicy.maximumRedeliveries=6

  

6. messagePrioritySupported

    Consumer端是否支持权重,默认为true,表示Consumer接受的所有消息都会按照权重排序,即权重较高的消息,优先被传递给 messageListener.onMessage()方法等。如果为false,则使用FIFO队列,消息达到Consumer之后将会加入到队列, 忽略权重属性,不过这在消息有重发的情况下,可能有少许影响。这个参数可以配合broker端“strictOrderDispath”达成消息权重排 序。[参见Priority]

 

//在brokerUrl中设置
tcp://localhost:61616?jms.messagePrioritySupported=true

 

7. priority

    消费者权重,这个权重和消息的权重不同,它用来标记Consumer消费消息的优先级,broker端将会对优先级较高的consumer,优先转发消息 (优先填充pending buffer),比如Consumer1的权重为10,Consumer2的权重为5,它们的prefetch(预获取消息的buffer尺寸)都是 10,那么当Broker端有12条消息,将会优先将Consumer1的buffer填充完毕(获取10条消息),Consumer2将会获得2条消 息。我们通常可以对网络良好、业务简单(比如selector更加简单)的Consumer设定较高的权重。参见ConsumerInfo类。

 

//在destinationUril中设定,默认所有的consumer权重都一样,为0
orderQueue?consumer.priority=10

 

    不过对于Topic而言,当指定了consumer的priority之后,还有一个可选的转发策略,用来优化消息传送优先级,理论上所有的 Subscripter(订阅者)都会收到相同的消息,但是我们在broker端的转发时机上,让优先级较高的订阅者先得到消息,尽管这看起来似乎没有什 么意义:

 

//对Queue无效
<policyEntry topic=">">  
    <dispatchPolicy>  
        <priorityDispatchPolicy />  
    </dispatchPolicy>  
</policyEntry>

 

 

8. optimizeAcknowledge/optimizeAcknowledgeTimeOut

    可优化ACK策略,这个是Consumer端最重要的调优参数之一。 ActiveMQ在Client端以及broker端,启用了大量的工作,用来优化消息确认(ACK),最大化消息消费效率。 optimizeAcknowledge表示是否开启“优化ACK选项”,当为true时,可以指定 optimizeAcknowledgeTimeOut数值用来约束ACK最大延迟确认的时间,我们通过optimizeAck,可以实现可靠的批量消息 确认。具体参看[ActiveMQ消息确认机制]      

 

//brokerUrl中,默认开启,timeout为300
tcp://localhost:61616?jms.optimizeAcknowledge=true&jms.optimizeAcknowledgeTimeOut=3000

 

    如果我们的consumer足够快,且服务器端的消息足够多,且指定了合适的prefethSize,我们可以将timeout时间设置的稍微大一些。建议不要关闭此选项。

 

9. prefetchSize

    预获取消息数量,重要的调优参数之一,当Consumer活跃时,broker将会批量发送prefetchSize条消息给 Consumer,consumer也可以配合optimizeAcknowledge来批量确认它们;批量传送,极大的提高了网络传输效率,此值默认为 1000。通过上述,我们对broker消息转发机制的了解,可以知道,broker端将会根据consumer指定的prefetchSize来决定 pendingBuffer的大小,prefetchSize越大,broker批量发送的消息就回越多,如果消费者消费速度较快,再配合 optimizeAck,这将是相对完美的消息传送方案。

 

    不过,prefetchSize也会带来一定的问题,在Queue中(Topic中没有效果),broker将使用“轮询”的方式来平衡多个消费者之间的 消息传送数量。如果消费者消费速度较慢,而且prefetchSize较大,这将不利于消息量在多个消费者之间平衡。通常情况下,如果consumer数 量较多,或者消费速度较慢,或者消息量较少时,我们设定prefetchSize为较小的值。

 

//在destinationUri中指定,默认为1000
//也可以在Topic中使用,可以优化ACK策略
orderQueue?customer.prefetchSize=100

 

    对于Consumer而言,支持同步消费(receive)和异步侦听(messageListener),这两种方式都经常用。无论何种情 况,prefetchSize都可以提供有效的优化。需要注意,如果开发者使用messageListener方式异步侦听消息,将不能设定 prefetchSize <= 0的任何值。如果使用receive方式,且prefetchSize = 0时,将触发Client端使用Pull机制,即每次receive调用,都会向Broker端发送Pull指令,如果broker端有消息才会转发,在 这种情况下,Broker不会主动Push消息给client。

 

10. noLocal

    consumer是否接受本地消息,所谓Local,就是和Comsumer共享Connection实例的Producer发送的消息。在此需要提醒, 当客户端创建connection时都会生成一个全局唯一的connectionId,当consumer、producer创建时都会向broker发 送标识信息,比如consumer创建时会向broker发送一个ConsumerInfo数据包(数据包中包含ConsumerId,唯一标记一个 consumer,consumerId中即包含ConnectionId),Producer同样在创建时发送携带有ProducerId的 ProducerInfo数据包;最重要的是,Producer发送消息时,所有的消息都封装了MessageId属性,这个属性中将携带 ProducerId,那么我们就很方便根据这些信息判断是否为Local,如果消息被持久化,connectionId也将随之。

 

    

//默认为false,如果你通过此方式设定了nolocal属性
//那么在创建consumer时指定的值将会被忽略。
orderQueue?customer.noLocal=false

 

  • session.createConsumer(topic,null,true)

    我们通常在Topic中使用noLocal,以忽略本地的消息通知等。

 

11. exclusiveConsumer

    “排他消费者”,此参数仅对Queue有效。即如果Queue通道中,有多个consumer同时活跃时只会有一个consumer能够获取消息,对于 broker而言,如果Queue中,有任意一个Consumer是“排他的”,那么消息只会转发给“exclusiveConsumer=true”的 消费者;如果全部的消费者都是“排他的”,那么最新创建的consumer将会获取消息。我们通常在分布式环境中,为了避免对某些重要数据并发操作时使用 此特性,比如:订单中心修改订单状态(来自各个系统的消息,都想修改订单状态,但是它们必须串行操作)。

 

//在brokerUrl中指定,默认值为false,对connectionFactory下所有的Queue有效
tcp://localhost:61616?jms.exclusiveConsumer=true

 

    我们也可以通过api的方式,为特定的Queue下的消费者设定此值,记住,只要全局中有任何一个Consumer设定了exclusive=true, 都会导致整个Queue的所有的Consumer为“排他的”,直到此Consumer失效且其他Consumer中没有“排他的”时,才会解除排他性。

 

//在destinationUri中设定,只对当前Queue有效。
orderQueue?consumer.exclusive=true

 

     不过,此处还有一个小小的陷阱,如果exclusive和selector(消息选择器)同时工作时,将会是什么结果呢??比如两个Consumer, 其中一个为exclusive,但是它们的selector不同,究竟两个Consumer是否都能够获取消息呢?事实上,尽管通过selector匹配 消息通过,但是如果使用了exclusive,且当前的“排他消费者”不是自己,那么消息仍然不能传送到client端。但是,如果两个consumer 的selector一样,那么最终将会有“排他消费者”获取消息。由此可见,selector和exclusive在某些方面具有“同样的效果”,如果所 有consumer的selector都不同(且选择范围不具有交集),且使用了exclusive,将会导致一部分消息被“积压”而无法消费,我们应该 避免这种情况的发生。(源码请参见: QueueDispatchSelector)

 

    此外,broker端还支持一个参数“allConsumersExclusiveByDefault”,如果此参数为true,那么当前Queue中的所有消费者默认为exclusive。

 

<!-- 只对queue有效,默认为false -->
<policyEntry queue="orderQueue" allConsumersExclusiveByDefault="true" />

 

12. retroactive

    “可追溯”消费者,只对Topic有效,如果consumer是可追溯的,那么它可以获取实例创建之前的消息。通常而言,订阅者不可能获取实例创建之前的 消息,因为broker根本不知道它的存在。对于broker而言,如果一个Topic通道创建,且有发布者发布消息(Publisher),那么 broker将会在内存中(非持久化)或者磁盘中(持久化)保存已经发布的消息,直到所有的订阅者都消费者,才会清除原始消息内容。那么 retroactive类型的订阅者,就可以获取这些原本不属于自己但broker上还保存的旧消息,就像我们订阅一种Feed,可以立即获取旧的内容列 表一样。如果此订阅者不是durable(耐久的),它可以获取最近发布的一些消息;如果是durable,它可以获取存储器中尚未删除的所有的旧消息。 [下文会详细介绍Topic的数据转发模型]

 

//在destinationUrl中设置,默认为false
feedTopic?consumer.retroactive=true

 

    在broker端,可以配置当前Topic默认为“可追溯的”,不过Topic并不会在此种情况下额外的保存消息,只不过表示订阅者默认都是可追溯的而已。

 

<!-- 只对topic有效,默认为false -->
<policyEntry topic="feedTopic" alwaysRetroactive="true" />

 

 

13. selector(选择器)

    在Consumer中使用选择器可以帮助broker过滤消息,就像我们使用在RDBMS中使用where字句过滤数据一样,具体selector的写 法,此处不做详细介绍。当Consumer创建之后,将会把selector信息传递给broker,此后再consumer的整个生命周期中,都将有 效,不过一旦指定selector,重复设定selector将不会有效,除非关闭consumer并重建实例。

 

    在Topic和Queue使用selector的时机,有所不同。

    对于Queue而言,broker在dispatch每条消息时,都会遍历整个消费者列表,并匹配selector表达式,一旦匹配成功则将消息发送给 Consumer,如果所有的selector都无法匹配,消息将沉积。注意,这些沉积的消息将会在每次pageIn时都会被加载而且也会在内存中不断叠 加(直到过期),将会对Queue的转发效率带来很大的危险,如果你发现Queue的消息大量积压(undeque),你应该检测是否与selector 有关。如果你使用了selector,你一定要让全局中所有的selector覆盖所有的消息,或者至少有一个没有selector的consumer。 (参看源码:Queue,QueueDispatchSelector)

 

    对于Topic而言,似乎有些不同,主要是在selector应用的时机上;如果订阅者是durable(耐久的),那么订阅者的ClientId + selector信息都会被持久化保存(参见TopicMessageStore),此后即使订阅者离线,符合selector的消息,仍然会为它创建消 息副本(为当前订阅者ID 与messageId的列表关系,但实际消息只有一条),如果selector不匹配,存储器将不会为此订阅者创建消息副本,也意味着当此订阅者上线后将 不会感知到任何事情,就像那些消息从来都没有出现过一样;不过在broker转发消息给订阅者之前,仍然会使用selector匹配;因为订阅者可以修改 selector,为了避免现有的消息副本不适合新的selector,将会在消息发送给订阅者时也使用selector匹配,对于不匹配 selector的消息,此时将会按照“已消费”来处理,它们也不会发送给订阅者。此处需要提醒,创建durable类型的订阅者时,需要在 connection上指定ClientId,全局中同一个Topic上所有的ClientId都不能相同,且同一个Connection上,不能创建多 个duarable订阅者。

 

    如果订阅者是非durable(即为temporary),也就不存在创建消息副本的情况,那么将会在消息转发给订阅者之前检测即可。

    (源码参见:Topic,DurableTopicSubscription,TopicMessageStore,RoundRobinDispatchPolicy等)

 

14. group消息分组

    JMS规范中并没有提供消息分组的特性,不过ActiveMQ支持了此特性;Producer在发送消息时为多条消息设定group,它们将会被同一个 consumer消费,这也是实现消息黏性的手段,任何时候一个group的消息只会发送给同一个Consumer,一个Consumer同时也只能消费 一个group直到黏性解除。此特性只会对Queue有效。在broker有一个groupOwnerMap用来保存groupId与 ConsumerId之间的关系,在消息转发时,如果消息中指定了groupId,就会使用此map检测是否已经有消费者持有了此group,如果此 group的消费者是自己,那么消息会继续转发;否则,消息继续被滞留。

 

    Message中可以通过指定隐藏属性JMSXGroupID来实现:

    1) message.setStringProperty("JMSXGroupID","order_1001_group")

    2) 也可以使用activeMQMessage.setGroupID(string)。

 

    我们可以通过JMSXGroupSeq来获取当前消息在group的序 号:message.getIntProperty("JMSXGroupSeq"),也可以使用 activeMQMessage.setGroupSeq(int),此属性并不会决定消息在broker端排序,通常有Producer端来决定此值, 仅仅是序号而已;如果此值为-1,表示此消息为group中最后一条消息,broker将会解除group与consumer的黏性关系,此后此 group的消息将会被重新选择消费者,即从groupOwnerMap中移除groupId键,同时也意味着这个consumer可以切换到其他 group上;如果此值为1表示,表示group重置(或者说重定向,阻断),当broker端在给某个consumer转发消息时遇到groupSeq = 1的消息,都会立即解除此此consumer与其他group的黏性,并立即与此group绑定,即将groupId与当前Consumer信息加入到 groupOwnerMap中。

 

    我们通常使用-1来表示一个group中的所有序列已经发送完毕,Consumer端可以获取根据这个属性值做批量操作的提交等,例如,consumer 接受了一个group中的消息,先把消息cache起来,当接收到groupSeq = -1 时,将cache中的数据批量写入文件,不过需要注意,consumer不能保证一定会收到-1的消息。groupSeq = 1是一个很特殊的值,有可能当前consumer整在接收group1的消息,此时如果group2发送了groupSeq = 1的消息,将会阻断consumer对group1的消费,转而消费group2的消息,所以consumer端在使用group时,需要对每个消息检测 groupId是否发证了“重置”,而不是一味的等待groupSeq = -1。当你决定此group信息已经发送完毕时,一定要发送seq = -1;如果你决定重新开启group时,需要发送seq = 1消息。

 

    如果group中的消息有部分发送给某个consumer(此consumer可能未能完全消费group而失效了),那么剩余的消息可能被转发到其他 consumer中,我们可以通过JMSXGroupFirstForConsumer(boolean 类型属性)来检测当前consumer是否为group的第一个消费者。如果使用了groupSeq = 1重置了group,那么也会强制将此属性设置为true。

 

    如果开发者的环境中,分组消息和普通消息混合使用,需要注意一些问题:consumer可能同时会接收到普通消息和分组消息,而并非同一时间只会接受分组 消息,不能过度依赖group的稳定性。也期望producer端不要随意创建groupId,而是使用相对固定的groupId,结合groupSeq = -1来实现group与consumer解耦;如果producer客户端较多,有可能同时并行的group较多,我们可以通过设定一个参数来实现 group在consumer的相对平稳:

 

 

//broker在转发消息之前需要等待足够多的consumer活跃或者等待超时后,才会发送消息
<policyEntry queue="orderQueue" consumersBeforeDispatchStarts="2" timeBeforeDispatchStarts="10000"/>

 

 

 

    不过似乎还有些疑惑: consumer正在消费group中的消息,如果此group中某条消息不符合consumer的selector,那么这条消息会发送到此 consumer吗?会发送给其他consumer吗? 事实上,selector将会首先发挥效果,如果消息没有匹配consumer的selector,即使它的groupId与此consumer有黏性, 消息也不会发送到此consumer;这些信息将会被积压,直到consumer失效,或者group被重置,那么此后这些积压的消息会转发给其他 consumer;这也意味着group中的信息被“拆分”了,因为selector的问题,这些消息被“分段”后先后发给了不同的consumer。 (参见Queue.doActualDispatch())

 

     

15. Consumer端消息接收模型



  

    

 

16. 慢速消费者(Slow Consumer)

     慢速,是相对于producer而言;简单来说,producer不断产生新的消息,broker端在内存中已经积压的足够多(比如 cacheLimit已满),但是在转发给某个consumer时,发现此consumer仍然有大量的消息尚没有消费(ACK),broker会认为此 consumer是慢速的。在Queue中,如果已发送(dispatched)但没有消费(unAck)的消息条数 > prefetchSize时,此consumer被标记为Slow。在Topic中,如果cacheLimit已满,但是需要向此订阅者发送的消息量 > prefetchSize时,此订阅者被标记为Slow。简单描述为: 快速的producer生产的消息,不能被消费者及时的消费,而导致在broker端积压。

 

    通常,慢速消费者会给整个broker带来潜在的危险,积压消息可能耗尽broker内存,也可能会导致消息不断的从文件pageIn到内存然后swap 到文件中(如果使用了临时文件),而消耗磁盘IO,最重要的一点是,它们还会拖累Producer,导致producer端的阻塞而降低消息生产的效率, 从而牵连那些快速消费者也无法获取充足的消息。所以,我们需要根据应用需要,对慢速消费者以及可能产生的潜在危险做一些必要的容错。

 

    如果你发现了某个通道中有慢速消费者,你要么人为的让producer“减速”,或者增加consumer的数量。

 

    慢速消费者,在TemporaryTopic中会带来很多问题;对于TemporaryTopic而言,直到所有的订阅者都消费了消息实体,才会将消息从 内存中删除(durable),有此可见,如果活跃的订阅者中,有一个是慢速的,就会导致消息在内存中积压而无法删除,当内存(cacheLimit)达 到阀值时,会阻塞producer继续发送消息(Flow Control策略默认开启),此时那些快速的消费者无法继续获取新的消息,就这么一个"猪一样的队友",最终导致整个TemporaryTopic的消 息流动性低效。 

 

    当Topic、Queue中非持久化消息的存储,也会遇到上述问题。如果开启了Flow Control,内存已满的情况下,则会阻塞通道中所有的Producer(阻塞主线程,producer底层的Transport将会被block)继 续发送消息(waitForSpace),直到足够的message被消费且从内存中移除;如果支持sendFailIfNoSpace,则会在内存不足 时,向producer发送error信息,producer可以捕获异常做重试操作(不会block底层的transport)。如果没有开启Flow Control,消息将会被直接转发(也不会检测内存使用情况),具体的情况将有底层存储器、系统内存(systemUsage)、JVM来决定,如果 systemUsage满足条件(所有的通道共享内存)则正常,否则waitForSpace,此时可能会有OOM的潜在危险。所以通常建议开启Flow Control,以及为每个通道设定合适的memoryLimit。

 

 

<!-- 每个队列,内存使用上限为256M,userCache表示持久化的消息也可以缓存以提高转发效率 -->
<policyEntry queue=">" producerFlowControl="true" memoryLimit="256mb" useCache="false">

<systemUsage>  
 <systemUsage sendFailIfNoSpace="true">  
   <memoryUsage>  
     <memoryUsage limit="8gb"/><!-- 所有的通道缓存消息的总内存大小 -->  
   </memoryUsage>  
 </systemUsage>  
</systemUsage>

 

 

    对于Topic、Queue中持久化消息,我们就无需担心过多,只要磁盘足够即可。这些消息将会在持久化成功之后,从内存中移除(useCache?),它们不会有额外的性能影响。

 

    仅仅通过限制内存和Flow Control并不能保证整体的消息吞吐能力,我们需要做点什么!

 

    1) 关闭Slow Consumer

 

 

<policyEntry queue=">" producerFlowControl="true" memoryLimit="512mb">             
  <slowConsumerStrategy>
	<abortSlowConsumerStrategy abortConnection="false"/>
  </slowConsumerStrategy>
</policyEntry>

 

    brorker端一旦发现slow consumer,就将它注册到慢速消费者列表中,此后将有额外的线程扫描并关闭它们,其中abortConnection表示,是否关闭底层的 transport,默认为false,此时将会通过transport向client端发送一个指令(其中包括consumerId),当client 端(Session)接收之后,将会调用consumer.close()方法;如果此值为true,将会导致底层的transport链接被关闭,这是 很粗暴的办法,不过如果client端多个consumer共享一个connection的话,会导致所有的consumer被关闭,还是那句话:猪一样 的队友,害了整个团队。

 

    2) 抛弃旧消息(仅对Topic有效)

 

 

<policyEntry topic=">" producerFlowControl="true" memoryLimit="512mb">             
	<pendingMessageLimitStrategy>
	  <!-- 对于慢速消费者,只保留最近100条未消费的消息,仅对topic有效 -->
	  <constantPendingMessageLimitStrategy limit="100"/>
	  <!-- 值保留 2.5 * prefetchSize条消息
	  <prefetchRatePendingMessageLimitStrategy multiplier="2.5"/>
	  -->
	</pendingMessageLimitStrategy>
</policyEntry>

 

     pendingMessageLimitStrategy仅对Topic有效,可以保证慢速消费者不会拖累太久。不过,如果滞留的消息大于了pendingMessageLimit,我们需要使用“移除策略”来移除多余的消息。

<policyEntry topic=">" producerFlowControl="true" memoryLimit="512mb">             
	<messageEvictionStrategy>
		<oldestMessageEvictionStrategy />
	</messageEvictionStrategy>
</policyEntry>

 

    topic支持支持3种移除策略:

    (1) oldestMessageEvictionStrategy表示移除最旧的消息

    (2) uniquePropertyMessageEvictionStrategy表示移除根据属性值筛选消息并移除最旧的

    (3) OldestMessageWithLowestPriorityEvictionStrategy表示在旧消息中移除权重最低的。

 

    3) 写入临时文件

 

<policyEntry queue=">" producerFlowControl="true" memoryLimit="512mb">             
	<pendingQueuePolicy>
		<storeCursor />
    </pendingQueuePolicy>
</policyEntry>

 

    对于Queue而言,支持storeCursor,vmQueueCursor , fileQueueCursor。其中storeCursor表示将非持久化消息写入store(根据配置的storeAdapter决定,比如使用 kahadb等),vmQueueCursor基于内存,fileQueueCursor表示将数据写入本地临时文件。

 

<policyEntry topic=">" producerFlowControl="false" memoryLimit="10mb">
	<pendingSubscriberPolicy>
		<!-- 对于非耐久的订阅者,非持久化消息: vmCursor,fileCursor -->
		 <fileCursor/>
	</pendingSubscriberPolicy>
	<pendingDurableSubscriberPolicy>
		<!-- 对于耐久的订阅者,非持久化消息 -->
		<!-- storeDurableSubscriberCursor -->
		<!-- vmDurableCursor -->
		<!-- fileDurableSubscriberCursor -->
		<storeDurableSubscriberCursor/>
	</pendingDurableSubscriberPolicy>
</policyEntry>

 

    对于Topic而言,我们也可以根据订阅者的类型,来决定如果处理那些滞留的非持久化消息。

 


© 著作权归作者所有

上一篇: activemq特性
泡海椒
粉丝 11
博文 282
码字总数 288344
作品 0
成都
程序员
私信 提问
activemq重启时,Consumer进程必须重新加载才能收到消息

情况是这样的,如果先启动activemq,然后再启动Producer和Consumer,消息可以收到。此时,如果命令行关闭activemq,Producer立马会报错,这很正常,因为connection连不上了嘛。然后Consumer和l...

Nykko
2016/01/06
1K
2
ActiveMQ在C#中的应用

ActiveMQ是个好东东,不必多说。ActiveMQ提供多种语言支持,如Java, C, C++, C#, Ruby, Perl, Python, PHP等。由于我在windows下开发GUI,比较关心C++和C#,其中C#的ActiveMQ很简单,Apache提...

postdep
2015/08/24
0
0
通过 ActiveMQ 演示如何使用 JMS API

消息机制是软件组件和应用程序之间通讯的一种方法,一个消息系统是一个点对点的服务:消息客户端可以发送消息也可以接收消息,每一个客户端连接到一个消息代理,这个代理提供了消息的创建、发...

tsl0922
2012/11/27
1K
0
ActiveMQ专题2: 持久化

AMQ的持久化问题 前言 前面一篇AMQ专题中,我们发现对于Topic这种类型的消息,即使将deliveryMode设置为持久化,只要生产者在消费者之前启动。消息生产者发布的消息还是会丢失。这是符合JMS...

槟城码农
2018/08/30
0
0
通过 ActiveMQ 演示如何使用 JMS API

消息机制是软件组件和应用程序之间通讯的一种方法,一个消息系统是一个点对点的服务:消息客户端可以发送消息也可以接收消息,每一个客户端连接到一个消息代理,这个代理提供了消息的创建、发...

红薯
2012/06/13
3.5K
6

没有更多内容

加载失败,请刷新页面

加载更多

利用mybatis generator生成实体类、Mapper接口以及对应的XML文件

项目中通常会遇到数据的持久化,如果是采用mybatis的orm,就会涉及到生成xml的问题,刚好mybatis官网提供了这么个插件MyBatis Generator,效果简直是棒呆。 1. 首先需要在build.gradle文件中...

啊哈关关
今天
2
0
SpringSocial相关的知识点

使用SprigSocial开发第三方登录 核心类 ServiceProvider(AbstractOauth2ServiceProvider):主要负责实现server提供商(例如QQ,微信等共有的东西),默认实现类是AbstractOauth2ServiceProvider...

chendom
今天
4
0
Java并发之AQS详解

一、概述   谈到并发,不得不谈ReentrantLock;而谈到ReentrantLock,不得不谈AbstractQueuedSynchronizer(AQS)!   类如其名,抽象的队列式的同步器,AQS定义了一套多线程访问共享资源...

群星纪元
昨天
4
0
Fabric-sdk-java最新教程

Fabric Java SDK是Fabric区块链官方提供的用于Java应用开发的SDK,全称为Fabric-sdk-java,网上可用资料不多,本文列出了精心整理的针对Fabric Java SDK的最新精选教程。 如果希望快速掌握F...

汇智网教程
昨天
3
0
react 子组件监听props 变化

componentWillReceiveProps //已经被废弃 getDerivedStateFromProps// 推荐使用//如果条件不存在必须要返回null static getDerivedStateFromProps(props, current_stat...

一箭落旄头
昨天
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部