ActiveMQ之Consumer

2021-01-05 本文已影响0人爱健身的兔子

1 Exclusive Consume（独占消费）

默认Queue中的消息是按照顺序被分发到consumer的，然而，当你有多个consumers同时从相同的queue中提取消息时，你将失去这个保证。因为这些消息是被多个线程并发的处理。 ActiveMQ从4.x版本开始支持Exclusive Consumer。Broker会从多个Consumers中挑选一个consumer来处理queue中所有的消息，从而保证了消息的有序处理。如果这个consumer失效，那么broker会自动切换到其他的consumer。可以通过destination options来创建一个Exclusive Consumer,如下：

private static final String queueName = "myQueue?consumer.exclusive=true";

说明：

这个会独占这个队列频道，所有的消息都将发发到这个连接上。在多线程环境下仍然是就这一个连接可以获取到消息。

2 Consumer Dispatche Async（消息异步分发）

在activemq4.0以后，你可以选择broker同步或异步的把消息分发给消费者。可以设置dispatchAsync属性，默认是true,通常情况下这是最佳的。你也可以通过如下几种方式修改：

在ConnectionFactory层设置

ActiveMQConnectionFactory.setDispatchAsync(false);

在Connection上设置,这个设置将会覆盖ConnectionFactory上的设置

ActiveMQConnetion.setDispatchAsync(false);

在Consumer上设置

　　queue = new ActiveMQQueue("TEST.QUEUE?consumer.dispatchAsync=false");
　　consumer = session.createConsumer(queue);

关闭异步分发

<transportConnector name="openwire" uri="tcp://0.0.0.0:61616" disableAsyncDispatch="true"/>

3 Consumer Priority（消息优先级）

当多个Consumer消费同一个队列的时候，可以为Consumer设置优先级来消费。

Consumer的Priority的划分为”0~127”个级别，127是最高的级别，0是最低的也是ActiveMQ默认的。这种配置可以让Broker根据consumer的优先级来发送消息到较高的优先级的Consumer上，如果某个较高的Consumer的消息消费慢，则Broker会把消息发送到仅次于它优先级的Consumer上。

queue = new ActiveMQQueue("TEST.QUEUE?consumer.priority=10");
consumer = session.createConsumer(queue);

4 Manage Durable Subscribers（管理持久订阅）

持久订阅，保证了消费者离线之后，再次进入系统，不会错过消息，但是这也会造成消息的堆积。可以通过设置消息的过期时间来控制，然后定期检查过期消息并删除：

<policyEntry topic=">" expireMessagesPeriod="300000"/>

从5.6开始，可以对不活跃的持久化订阅进行清除。如下：

<broker name="localhost" 
        offlineDurableSubscriberTimeout="86400000" 
            offlineDurableSubscriberTaskSchedule="3600000">

说明：

offlineDurableSubscriberTimeout：离线多长时间就过期删除，缺省是-1，就是不删除。 offlineDurableSubscriberTaskSchedule：多长时间检查一次，缺省300000，单位毫秒。

5 Message Groups（消息分组）

Message Goups就是对消息分组，它是Exclusive Consumer功能的增强。逻辑上Message Groups可以看成是一种并发的 Exclusive Consumer。跟所有的消息都由唯一的consumer处理不同，JMS消息属性的JMSXGroupID用来区分message group。Message Group特性保证所有具有相同 JMSXGroupID 的消息都会被分发到相同的consumer(只要这个consumer保持active) 。
另一方面，Message Groups 特性也是一种负载均衡的机制。在一个消息被分发到consumer之前，broker首先检查消息 JMSXGroupID 属性。如果存在，那么broker会检查是否有某个consumer拥有这个message group.如果没有，那么broker会选择一个consumer,并将它关联到这个message group.此后，这个consumer会接收到这个message group的所有消息，直到：

consumer被关闭。
Message group被关闭（通过发送一个消息，并设置这个消息的 JMSXGroupSeq 为-1）。

创建一个Message Groups，只需要在message对象上设置属性即可，如下：

message.setStringProperty("JMSXGroupID","GroupA");

关闭一个Message Groups，只需要在message对象上设置属性即可，如下：

　message.setStringProperty("JMSXGroupID","GroupA");
　message.setIntProperty("JMSXGroupSeq",-1);

6 Message Selectors（消息选择器）

JMS Selectors 用在获取消息的时候，可以基于消息属性和 Xpath 语法对消息进行过滤。JMS Selectors有SQL92语义定义。以下是个Selectors的例子：

//创建消费者的时候指定 Selector
MessageConsumer consumer = session.createConsumer(destination, "JMSXGroupID='GroupA'");

注意：

JMS Selectors表达式中，可以使用IN, NOT IN, LIKE等。
需要注意的是，JMS Selectors 表达式中的日期和时间需要使用标准的Long型毫秒值。
表达式中的属性不会自动进行类型转换。
Message Groups虽然可以保证具有相同的message group的消息会被唯一的consumer顺序处理，但是却不能确定被哪个consumer处理。

7 Redelivery Policy（重传策略）

当Consumer在接收到消息时，由于以下原因没有返回Ack信息，就会导致消息的重新投递。

Client用了transactions，且在Session中调用了rollback()。
Client用了transactions，且在调用commit()之前关闭。
Client在CLIENT_ACKNOWLEDGE的传递模式下，在session中调用了recover()。
Client在超时回复Ack。

可以通过设置ActiveMQConnectionFactory和ActiveMQConnection来定制想要的重传策略，可用的Redelivery属性如下：

property	default value	description
collisionAvoidanceFactor	0.15	设置防止冲突范围的正负百分比，只有启用了useCollisionAvoidance参数时才生效。也就是在延迟时间上再加一个时间波动范围。
maximumRedeliveries	6	最大重传次数，达到最大重传次数后抛出异常。为-1时不限制次数，为0时表示不进行重传。
maximumRedeliveryDelay	-1	传送延迟，旨在useExpoentialBackOff为true时有效（5.5之后），假设首次重间隔为10ms,倍数为2，那么第二次重连时间间隔为20ms,第三次重连时间间隔为40ms,当重连时间间隔大于最大重连时间间隔时，以后每次重连时间间隔都为最大重连时间间隔。
initialRedeliveryDelay	1000L	初始重发延迟时间。
redeliveryDelay	1000L	重发延迟时间，当initialRedeliveryDelay=0时生效。
useCollisionAvoidance	false	启用防止冲突功能。
useExponentialBackOff	false	启用指数倍数递增的方式增加延迟时间。
backOffMultiplier	5	重连时间间隔递增倍数，只有值大于1和启用useExponentialBackOff参数时才生效。

示例

// 1创建ConnectionFactory
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);

RedeliveryPolicy policy = new RedeliveryPolicy();
policy.setMaximumRedeliveries(3);// 重传次数
policy.setInitialRedeliveryDelay(2 * 1000);
connectionFactory.setRedeliveryPolicy(policy);

8 Slow Consumer Handling（慢消费处理）

8.1 Prefetch机制

ActiveMQ通过Prefetch机制来提供性能，在客户端得内存里缓存一定数量得消息。缓存消息得数量由prefetch limit来控制。当某个 consumer 的 prefetch buffer 已经达到上限或者消息缓冲区满了，那么broker不会再向consumer分发消息，直到consumer像broker发送消息的确认，确认后的消息将会从缓存中去掉。ActiveMQ默认预取数量如下：

转发模式	队列类型	prefetchSize
PERSISTENT	Queue	1000
NON_PERSISTENT	Queue	1000
PERSISTENT	Topic	100
NON_PERSISTENT	Topic	32766

注意：

如果预取数量设置为1，将会导致消息一条一条的推送。如果预取数量被设置为0，将会导致关闭Broker的推送功能，需要消费者主动拉取数据。

prefetchk可以通过如下方式设置：

通过URL

tcp://localhost:61616?jms.prefetchPolicy.all=50
tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch=1
或
queue = new ActiveMQQueue("TEST.QUEUE?consumer.prefetchSize=10");

通过ActiveMQPrefetchPolicy策略对象修改

ActiveMQPrefetchPolicy prefetchPolicy = connectionFactory.getPrefetchPolicy();
prefetchPolicy.setQueuePrefetch(200);

通过Properties属性设置

Properties properties = connectionFactory.getProperties();
properties.setProperty("prefetchPolicy.queuePrefetch","1000");

8.2 慢消费处理

生产者在发送消息，Broker在将消息转发给消费者时，如果发现其内部有大量消息没有消费完成，那么Broker就会认为该消费者是慢消费者。在队列模式下，如果已发送，但没有确认的消息数量大于 prefetchSize，则消费者会被标记为 Slow。在主题模式下，如果cacheLimit已满，但是向主题的订阅者要发送的消息大于 prefetchSize，那么订阅者将被标记为 Slow。

由于慢消费者会导致消息堆积，消耗内存，从而导致内存数据和磁盘文件不停交换，消耗磁盘IO。同时还会影响生产者生成消息的速率。ActiveMQ使用等待消息限制策略（Pending Message Limit Strategy）来解决这个问题，当超过这个上限后有新消息到来时将根据不同的策略抛弃。

等待消息限制

目前等待消息限制策略有以下两种：

Constant Pending Message Limit Strategy

limit 可以设置0， > 0, -1三种方式：0表示：不额外的增加其预存大小，> 0表示：在额外的增加其预存大小，-1表示：不增加预存也不丢弃旧的消息，这个策略使用常量限制，配置如下：

<constantPendingMessageLimitStrategy limit="50"/>

Prefetch Rate Pending Message LimitStrategy

这种策略是利用Consumer的之前的预存的大小乘以其倍数等于现在的预存大小。比如：

　<prefetchRatePendingMessageLimitStrategy multiplier="2.5"/>

注意：

在以上两种方式中，如果设置了0，意味着除了prefetch之外不再缓存消息，如果设置了-1意味着禁止丢弃消息。

消息丢弃

目前消息丢弃策略有三种：

oldestMessageEvictionStrategy：这个策略丢弃最旧的消息。
oldestMessageWithLowestPriorityEvictionStrategy：这个策略丢弃最旧的，而且具有最低优先级的消息。
uniquePropertyMessageEvictionStrategy:从5.6开始，可以根据自定义的属性来进行抛弃。<uniquePropertyMessageEvictionStrategy propertyName="STOCK"/>表示要抛弃属性名称为Stock的消息。

示例：

<policyEntry topic="PRICES.>"> 
            <!-- lets force old messages to be discarded for slow consumers -->  
            <pendingMessageLimitStrategy> 
              <constantPendingMessageLimitStrategy limit="10"/> 
            </pendingMessageLimitStrategy>  
            <!-- 10 seconds worth -->  
            <messageEvictionStrategy> 
              <uniquePropertyMessageEvictionStrategy propertyName="STOCK"/> 
            </messageEvictionStrategy> 
</policyEntry>

9 Subscription Recovery Policy（订阅恢复策略）

生产者在某个topic发送了多条消息后，这个时候非持久订阅者才订阅，那么它是不能获取之前生产者发送的信息的。或者，由于网络问题，非持久类型的消费者处于非活跃状态，无法接收到生产者发送的消息。使用消息恢复策略，可以解决上面的问题。ActiveMQ目前支持一个定时或固定大小的恢复缓冲区，在你连接到broker后，在一段时间内的消息会重新发送给订阅者。

Policy Name	Sample Configuration	Description
FixedSizedSubscriptionRecoveryPolicy	`<fixedSizedSubscriptionRecoveryPolicy maximumSize="1024"/>`	保留固定字节的消息。
FixedCountSubscriptionRecoveryPolicy	`<fixedCountSubscriptionRecoveryPolicy maximumSize="100"/>`	保留固定数量的消息。
LastImageSubscriptionRecoveryPolicy	`<lastImageSubscriptionRecoveryPolicy/>`	保留最后一条记录。
NoSubscriptionRecoveryPolicy	`<noSubscriptionRecoveryPolicy/>`	禁用回溯，这是默认配置。
QueryBasedSubscriptionRecoveryPolicy	`<queryBasedSubscriptionRecoveryPolicy query="JMSType = 'car' AND color = 'blue'"/>`	根据查询机制使用回溯。
TimedSubscriptionRecoveryPolicy	`<timedSubscriptionRecoveryPolicy recoverDuration="60000" />`	保留指定时间内的消息。
RetainedMessageSubscriptionRecoveryPolicy	`<retainedMessageSubscriptionRecoveryPolicy/>`	保留ActiveMQ.Retain属性值为true的最后1条消息。

示例：

<policyEntry topic=">">
    <subscriptionRecoveryPolicy>
        <timedSubscriptionRecoveryPolicy recoverDuration="60000"/>
    </subscriptionRecoveryPolicy>
</policyEntry>

注意：

需要设置retroactive属性为true。如下：

Topic topic = session.createTopic("TEST.TOPIC?consumer.retroactive=true");
MessageConsumer consumer = session.createConsumer(topic);

JMS学习(八)-ActiveMQ Consumer 使用 push 还是 pull 获取消息 - hapjin - 博客园
 ActiveMQ