Consumer高级特性

Posted qlqwjy

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Consumer高级特性相关的知识,希望对你有一定的参考价值。

  Queue队列的消息一般是按照顺序各个队列依次获取消息,每次获取一个。所以假设有两个队列queue1,queue2,发送的消息为1、2、3、4、5。则默认情况下queue1获取到的消息为1、3、5,queue2获取到的消息为2、4.

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";

  还可以给consumer设置优先级,以便针对网络情况进行优化,如下:

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

 

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

2. Consumer Dispatche Async    用于队列消息

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

ActiveMQConnectionFactory.setDispatchAsync(false);

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

ActiveMQConnetion.setDispatchAsync(false);

3. 在Consumer上设置

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

3. Consumer Priority    用于队列消息

JMS JMSPriority定义了十个消息优先级值,0是最低优先级,9是最高优先级,另外,客户端应当将0-4看作普通优先级,5-9看作加急优先级。

自定义Consumer Priority优先级。配置如下:

private static final String queueName = "myQueue?consumer.priority=10";

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

 

通过测试发现:只有消息级别相同的情况下才会平分队列的消息,如果存在一方的级别高于其他那么级别高的会获取全部的消息。

4. Manage Durable Subscribers  用于主题消息

消息持久化,保证了消费者离线之后,再次进入系统,不会错过消息,但是这也会消耗很多的资源,从5.6开始,可以对持久化进行如下管理:
  Removing inactive subscribers
我们还希望可以删除那些不活动的订阅者,如下:

<broker xmlns="http://activemq.apache.org/schema/core" schedulerSupport="true" offlineDurableSubscriberTimeout="30000" offlineDurableSubscriberTaskSchedule="30000" brokerName="brokerName" dataDirectory="${activemq.data}" schedulePeriodForDestinationPurge="1000">

1.offlineDurableSubscriberTimeout:离线多长时间就过期删除,缺省是-1,就是不删除。
2. 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的所有消息,直到:
  1. consumer被关闭
  2. Message group被关闭,通过发送一个消息,并设置这个消息的JMSXGroupSeq为-1
  创建一个Message Groups,只需要在message对象上设置属性即可,如下:

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

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

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

 

例如:生产者将序号1、2、3的设为1组,两个消费者接收:

生产者代码:

        for (int i = 0; i < 5; i++) {
            // 7.创建Message,有好多类型,这里用最简单的TextMessage
            TextMessage tms = session.createTextMessage("textMessage:" + i);

            // 设置附加属性
            tms.setStringProperty("str", "stringProperties" + i);
            tms.setJMSPriority(6);
            tms.setJMSReplyTo(temporaryQueue);

            // 设置消息组
            if (i == 1 || i == 2 || i == 3) {
                tms.setStringProperty("JMSXGroupID", "GroupA");
            }

            // 8.生产者发送消息
            producer.send(tms);
        }

结果:

消费者2:

2019-04-12 14:51:33 [cn.qlq.activemq.queue.MsgConsumer2]-[INFO] [C2]threadname ->main,接收消息:textMessage:1;属性stringProperties1
2019-04-12 14:51:33 [cn.qlq.activemq.queue.MsgConsumer2]-[INFO] [C2]threadname ->main,接收消息:textMessage:2;属性stringProperties2
2019-04-12 14:51:33 [cn.qlq.activemq.queue.MsgConsumer2]-[INFO] [C2]threadname ->main,接收消息:textMessage:3;属性stringProperties3

 

消费者1:

2019-04-12 14:51:33 [cn.qlq.activemq.queue.MsgConsumer]-[INFO] [C2]threadname ->main,接收消息:textMessage:0;属性stringProperties0
2019-04-12 14:51:33 [cn.qlq.activemq.queue.MsgConsumer]-[INFO] [C2]threadname ->main,接收消息:textMessage:4;属性stringProperties4

6. Message Selectors

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

        // 6.创建消费者consumer
        MessageConsumer consumer = session.createConsumer(destination, "sendTo=‘C2‘");

1. JMS Selectors表达式中,可以使用IN, NOT IN, LIKE等
2. 需要注意的是,JMS Selectors表达式中的日期和时间需要使用标准的Long型毫秒值。
3. 表达式中的属性不会自动进行类型转换,例如:

tms.setStringProperty("strVal", "2");

    那么此时“setVal > 1” 的结果就是会false
4. Message Groups虽然可以保证具有相同的message group的消息会被唯一的consumer顺序处理,但是却不能确定被哪个consumer处理,在某些情况下,Message Groups可以和JMS Selector一起工作。
  例如:设想三个consumers分别是A,B,C,你可以在producer中为消息设置三个message groups分别为“A","B","C"。然后令Consumer A使用JMSXGroupID=‘A‘作为selector,c和b也同理,这样就保证了message group A的消息只会被A处理,需要注意的是,这种做法有以下缺点:
  (1) producer必须直到当前正在运行的consumers,也就是说producer和consumer被耦合到一起。
  (2) 如果某个consumer失效,那么应该被这个consumer消费的消息将会一直被积压在broker上。

 

例如:设置一个sendTo属性标记发给哪个消费者

生产者:

            // 设置消息组
            if (i == 1 || i == 2 || i == 3) {
                tms.setStringProperty("sendTo", "C1");
            } else {
                tms.setStringProperty("sendTo", "C2");
            }

C1:

        // 6.创建消费者consumer
        MessageConsumer consumer = session.createConsumer(destination, "sendTo=‘C1‘");

C2:

        // 6.创建消费者consumer
        MessageConsumer consumer = session.createConsumer(destination, "sendTo=‘C2‘");

7.Redelivery Policy  重传策略

ActiveMQ在接收消息的Client有以下几种操作的时候,需要重新传递消息:
  1. Client用了transactions,且在Session中调用了rollback();
  2. Client用了transactions,且在调用commit()之前关闭。
  3. Client在CLIENT_ACKNOWLEDGE的传递模式下,在session中调用了recover();

可以通过设置ActiveMQConnectionFactory和ActiveMQConnection来定制想要的再次传送策略,可用的Redelivery属性如下:
  1). collisionAvoidanceFactor:设置防止冲突范围的正负百分比,只有启用了useCollisionAvoidance参数时才生效。也就是在延迟时间上再加一个时间波动范围。默认值是0.15
  2). maximumRedeliveries:最大重传次数,达到最大重传次数后抛出异常。为-1时不限制次数,为0时表示不进行重传。默认值为
  3) . maximumRedeliveryDelay:传送延迟,旨在useExpoentialBackOff为true时有效(5.5之后),假设首次重间隔为10ms,倍数为2,那么第二次重连时间间隔为20ms,第三次重连时间间隔为40ms,当重连时间间隔大于最大重连时间间隔时,以后每次重连时间间隔都为最大重连时间间隔。默认为-1.
  4). initialRedeliveryDelay:初始重发延迟时间,默认1000L
  5). redeliveryDelay:重发延迟时间,当initialRedeliveryDelay=0时生效,默认1000L
  6). useCollisionAvoidance:启用防止冲突功能,默认false
  7). useExponentialBackOff:启用指数倍数递增的方式增加延迟时间,默认false
  8). backOffMultiplier:重连时间间隔递增倍数,只有值大于1和启用useExponentialBackOff参数时才生效,默认是5;

例如:测试session中发生异常,重传3次,间隔2s重传(总共传4次)

 生产者代码

package cn.qlq.activemq.queue;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TemporaryQueue;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 生产消息
 */
public class MsgProducer {

    // 默认端口61616
    private static final String url = "tcp://localhost:61616/";
    private static final String queueName = "myQueue";
    private static Session session = null;

    public static void main(String[] args) throws JMSException {
        // 1创建ConnectionFactory
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
        // 2.由connectionFactory创建connection
        Connection connection = connectionFactory.createConnection();
        // 3.启动connection
        connection.start();
        // 4.创建Session===第一个参数是是否事务管理,第二个参数是应答模式
        session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        // 5.创建Destination(Queue继承Queue)
        Queue destination = session.createQueue(queueName);

        TemporaryQueue temporaryQueue = session.createTemporaryQueue();

        // 6.创建生产者producer
        MessageProducer producer = session.createProducer(destination);
        for (int i = 0; i < 2; i++) {
            // 7.创建Message,有好多类型,这里用最简单的TextMessage
            TextMessage tms = session.createTextMessage("textMessage:" + i);

            // 设置附加属性
            tms.setStringProperty("str", "stringProperties" + i);
            tms.setJMSPriority(6);
            tms.setJMSReplyTo(temporaryQueue);

            // 8.生产者发送消息
            producer.send(tms);
        }

        // 9.提交事务
        session.commit();

        // 10.关闭connection
        session.close();
        connection.close();
    }

}

 

消费者代码: (处理第二个消息的时间制造异常进行事务回滚操作,消息会重发三次,之后会进入死队列)

package cn.qlq.activemq.queue;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.RedeliveryPolicy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 消费消息
 * 
 * @author QiaoLiQiang
 * @time 2018年9月18日下午11:26:41
 */
public class MsgConsumer {

    // 默认端口61616
    private static final String url = "failover:(tcp://localhost:61616,tcp://localhost:61617)?randomize=true";
    private static final String queueName = "myQueue?consumer.priority=9";
    private static final Logger LOGGER = LoggerFactory.getLogger(MsgConsumer.class);

    public static void main(String[] args) throws JMSException, InterruptedException {
        // 1创建ConnectionFactory
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);

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

        // 2.由connectionFactory创建connection
        Connection connection = connectionFactory.createConnection();
        // 3.启动connection
        connection.start();
        // 4.创建Session===第一个参数是是否事务管理,第二个参数是应答模式
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        // 5.创建Destination(Queue继承Queue)
        Queue destination = session.createQueue(queueName);
        // 6.创建消费者consumer
        MessageConsumer consumer = session.createConsumer(destination);

        int i = 0;
        while (i < 50) {
            TextMessage textMessage = (TextMessage) consumer.receive();
            LOGGER.info("[C2]threadname ->{},接收消息:{}", Thread.currentThread().getName(),
                    textMessage.getText() + ";属性" + textMessage.getStringProperty("str"));

            // 模拟处理第二条消息的时候 异常
            try {
                if ("textMessage:1".equals(textMessage.getText())) {
                    int ee = 1 / 0;
                }
            } catch (Exception e) {
                session.rollback();
            } finally {
                i++;
                session.commit();
            }
        }

        // session.close();
        // connection.close();
    }
}

结果:

2019-04-12 16:59:46 [org.apache.activemq.transport.failover.FailoverTransport]-[INFO] Successfully connected to tcp://localhost:61616
2019-04-12 16:59:49 [cn.qlq.activemq.queue.MsgConsumer]-[INFO] [C2]threadname ->main,接收消息:textMessage:0;属性stringProperties0
2019-04-12 16:59:49 [cn.qlq.activemq.queue.MsgConsumer]-[INFO] [C2]threadname ->main,接收消息:textMessage:1;属性stringProperties1
2019-04-12 16:59:51 [cn.qlq.activemq.queue.MsgConsumer]-[INFO] [C2]threadname ->main,接收消息:textMessage:1;属性stringProperties1
2019-04-12 16:59:53 [cn.qlq.activemq.queue.MsgConsumer]-[INFO] [C2]threadname ->main,接收消息:textMessage:1;属性stringProperties1
2019-04-12 16:59:54 [cn.qlq.activemq.queue.MsgConsumer]-[INFO] [C2]threadname ->main,接收消息:textMessage:1;属性stringProperties1

 

并且该消息进入死队列

技术图片

7.1 死队列

  当消息试图被传递的次数超多配置中的maximumRedeliveries属性的值时,那么,broker会认定该消息是一个死消息,并会把该消息发送到死队列中。默认activeMQ中死队列被声明为”ActiveMQ.DLQ",所有不能消费的消息都被传递到该死队列中。你可以在activemq.xml中配置individualDeadLetterStrategy属性,示例如下:

<policyEntry queue= "> " >
    <deadLetterStrategy>
        <individualDeadLetterStrategy queuePrefix= "DLQ." useQueueForQueueMessages= "true" />
    </deadLetterStrategy>
</policyEntry>

7.2 自动删除过期消息

有时需要直接删除过期的消息而不需要发送到死队列中,可以使用属性processExpired=false来设置,示例如下:

<policyEntry queue= "> " >
<deadLetterStrategy>
<sharedDeadLetterStrategy processExpired= "false" />
</deadLetterStrategy>
</policyEntry>

7.3 存放非持久消息到死队列中

默认情况下,Activemq不会把非持久的死消息发送到死队列中。非持久性如果你想把非持久的消息发送到死队列中,需要设置属性processNonPersistent=“true”,示例如下:

<policyEntry queue= "> " >
<deadLetterStrategy>
<sharedDeadLetterStrategy processNonPersistent= "true" />
</deadLetterStrategy>
</policyEntry>

7.4 为每一个Destination配置一个Redelivery Policy

RedeliveryPolicy per Destination:在5.7之后,你可以为每一个Destination配置一个Redelivery Policy,示例如:

ActiveMQConnection connection ... // Create a connection
RedeliveryPolicy queuePolicy = new RedeliveryPolicy();
queuePolicy.setInitialRedeliveryDelay(0);
queuePolicy.setRedeliveryDelay(1000);
queuePolicy.setUseExponentialBackOff(false);
queuePolicy.setMaximumRedeliveries(2);

RedeliveryPolicy topicPolicy = new RedeliveryPolicy();
topicPolicy.setInitialRedeliveryDelay(0);
topicPolicy.setRedeliveryDelay(1000);
topicPolicy.setUseExponentialBackOff(false);
topicPolicy.setMaximumRedeliveries(3);
// Receive a message with the JMS API
RedeliveryPolicyMap map = connection.getRedeliveryPolicyMap();
map.put(new ActiveMQTopic(">"), topicPolicy);
map.put(new ActiveMQQueue(">"), queuePolicy);

 

8.Slow Consumer Handling 

1.  Prefetch机制:

  ActiveMQ通过Prefetch机制来提供性能,方式是在客户端得内存里可能缓存一定数量得消息。缓存消息得数量由prefetch limit来控制。当某个consumer的prefetch buffer已经达到上限,那么broker不会再向consumer分发消息,知道consumer像broker发送消息的确认,确认后的消息将会从缓存中去掉。
  可以通过在ActiveMQConnectionFactory或者ActiveMQConnection上设置ActiveMQPrefetchPolicy对象来配置prefetch policy。也可以通过connection options或destination options来配置。例如:

  tcp://localhost:61616?jms.prefetchPolicy.all=50
  tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch=1

或者

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

 

prefetch size缺省的值如下:

  persistent queues (default value: 1000)
  non-persistent queues (default value: 1000)
  persistent topics (default value: 100)
  non-persistent topics (default value: Short.MAX_VALUE - 1) 

2. 慢Consumer处理

  慢消费者会在非持久的topics上导致问题,一旦消息积压起来,会导致broker把大量消息保存到内存中,broker也会因此而变慢,目前,ActiveMQ使用Pending Message Limit Strategy来解决这个问题。除了prefetch buffer之外,你还要配置缓存消息的上限,超过这个上限之后,新消息到来时会丢弃旧的消息。
  通过在配置文件的destination map中配置pendingMessageLimitStrategy,可以为不同的topic message配置不同的策略。
Pending Message Limit Strategy(等待消息限制策略),目前有以下两种“
  1. Constant Pending Message Limit Strategy
    Limit 可以设置0, > 0, -1三种方式:0表示:不额外的增加其预存大小,> 0表示:在额外的增加其预存大小,-1表示:不增加预存也不丢弃旧的消息,这个策略使用常量限制,配置如下:

  <constantPendingMessageLimitStrategy limit="50"/>

  2. Prefetch Rate Pending Message LimitStrategy
    这种策略是利用Consumer的之前的预存的大小乘以其倍数等于现在的预存大小。比如:

  <prefetchRatePendingMessageLimitStrategy multiplier="2.5"/>

  3. 说明:在以上两种方式中,如果设置了0,意味着除了prefetch之外不再缓存消息,如果设置了-1意味着禁止丢弃消息。
  配置消息的丢弃策略,目前有三种方式:
  oldestMessageEvictionStrategy:这个策略丢弃最旧的消息。
  oldestMessageWithLowestPriorityEvictionStrategy: 这个策略丢弃最旧的,而且具有最低优先级的消息。  
  uniquePropertyMessageEvictionStrategy:从5.6开始,可以根据自定义的属性来进行抛弃,比如  <uniquePropertyMessageEvictionStrategy propertyName="STOCK"/>表示要抛弃属性名称为Stock的消息。

配置方式:

技术图片

 

以上是关于Consumer高级特性的主要内容,如果未能解决你的问题,请参考以下文章

ActiveMQ(19):Consumer高级特性之独有消费者(Exclusive Consumer)

ActiveMQ——Consumer高级特性

ActiveMQ(22):Consumer高级特性之消息分组(Message Groups)

ActiveMQ(20):Consumer高级特性之重新投递(Redelivery Policy)

ActiveMQ(21):Consumer高级特性之管理持久化订阅(Manage Durable Subscribers)

ActiveMQ(23):Consumer高级特性之Message dispatche asyncConsumer Priority与Message Selectors