Consumer高级特性

时间：2019-04-12 23:07:26 阅读：150 评论：0 收藏：0 [点我收藏+]

标签：scheduler pre produce commit 继承 led process toc 第一个

　　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高级特性

标签：scheduler pre produce commit 继承 led process toc 第一个

原文地址：https://www.cnblogs.com/qlqwjy/p/10695939.html

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行

Consumer高级特性

1.Exclusive Consume 用于队列消息

2. Consumer Dispatche Async 用于队列消息

3. Consumer Priority 用于队列消息

4. Manage Durable Subscribers 用于主题消息

5. Message Groups

6. Message Selectors

7.Redelivery Policy 重传策略

7.1 死队列

7.2 自动删除过期消息

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

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

8.Slow Consumer Handling

1. Prefetch机制：

2. 慢Consumer处理

1.Exclusive Consume　用于队列消息　

4. Manage Durable Subscribers　　用于主题消息

5. Message Groups　　

8.Slow Consumer Handling　

1.　　Prefetch机制：