首页 > 代码库 > ActiveMQ(23):Consumer高级特性之Message dispatche async、Consumer Priority与Message Selectors

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

一、Message dispatche async(消息异步分发)

在activemq4.0以后,你可以选择broker同步或异步的把消息分发给消费者。可以设置dispatchAsync 属性,默认是true,通常情况下这是最佳的。

修改:可以通过如下几种方式

 1:在ConnectionFactory层设置

  ((ActiveMQConnectionFactory)connectionFactory).setDispatchAsync(false);

 2:在Connection上设置

  这个设置将会覆盖ConnectionFactory上的设置

  ((ActiveMQConnection)connection).setDispatchAsync(false);

 3:在Consumer来设置

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

  consumer = session.createConsumer(queue);

二、Consumer Priority(消费者优先级)

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


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


如何定义Consumer Priority的优先级呢? 配置如下:

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

  consumer = session.createConsumer(queue);

三、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使用“JMXGroupID = ‘A’”作为selector。B和C也同理。这样就可以保证message group A的消息只被consumer A处理。需要注意的是,这种做法有以下缺点:

(1)producer必须知道当前正在运行的consumers,也就是说producer和consumer被耦合到一起。

(2)如果某个consumer失效,那么应该被这个consumer消费的消息将会一直被积压在broker上。


本文出自 “我爱大金子” 博客,请务必保留此出处http://1754966750.blog.51cto.com/7455444/1924922

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