zoukankan      html  css  js  c++  java
  • ActiveMQ学习笔记(20)----Consumer高级特性(二)

    1. Message Selectors

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

      consumer = session.createConsumer(destination, "JMSType='car' AND weight > 2500");

      1. JMS Selectors表达式中,可以使用IN, NOT IN, LIKE等

      2. 需要注意的是,JMS Selectors表达式中的日期和时间需要使用标准的Long型毫秒值。

      3. 表达式中的属性不会自动进行类型转换,例如:

        myMessage.setStringProperty("NumberOfOrders","2");

        那么此时“NumberOfOrders  > 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上。

    2. 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;

      在接收的client可以如下设置:

      ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("failover:(tcp://localhost:61616,tcp://localhost:61617)?randowize=false");

      RedeliveryPolicy policy = new RedeliveryPolicy();

      policy.setMaximumRedeliveries(3);

      cf.setRedeliveryPolicy(policy);

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

    <policyEntry queue=">">
        <dealLetterStrategy>
            <individualDeadLetterStategy queuePrefix="DLQ."
                useQueueForQueueMessage="true"/>
        </dealLetterStrategy>
    </policyEntry>

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

    <policyEntry queue=">">
        <dealLetterStrategy>
            <sharedDeadLetterStategy processExpired="false"/>
        </dealLetterStrategy>
    </policyEntry>

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

    <policyEntry queue=">">
        <dealLetterStrategy>
            <sharedDeadLetterStategy processNonPersistent="true"/>
        </dealLetterStrategy>
    </policyEntry>

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

    3. Slow Consumer Handling

      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)

      慢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的消息。

      配置方式:

      

      

  • 相关阅读:
    关于文件上传组件国内外完美解决方案的调查
    WIN10 评估版 查看过期时间
    Using Android Phone to recover SD card formatted with DD command under linux
    JavaScript 字符串与数组互转,并保持数据去重、排序功能
    国内基于浏览器的在线截屏插件方案汇总分析
    VC6到VC2010,项目迁移错误
    CentOS 7 Install Adobe Flash Player
    地铁图快速寻路算法
    windows的cmd下的find命令比bash(win10下的Ubuntu的bash)下的grep比较
    APUE习题5.x
  • 原文地址:https://www.cnblogs.com/Eternally-dream/p/9916086.html
Copyright © 2011-2022 走看看