ActiveMQ的消息过滤器

1、消息过滤器的简介

      消息选择器的用法
      MessageConsumer是一个Session创建的对象,用来从Destination接收消息


      关于消息选择器
      MessageConsumer createConsumer( Destination destination, String messageSelector )
      MessageConsumer createConsumer( Destination destination, String messageSelector, boolean noLocal )

      其中,messageSelector为消息选择器; 
      noLocal标志默认为false,当设置为true时,限制消费者只能接收和自己相同的连接(Connection)所发布的消息,此标志只适用于主题,不适用于队列。

      public final String SELECTOR="JMS_TYPE='MY_TAG1'" ; 
      选择器检查传入消息的JMS_TYPE的属性,并确定这个属性的值是否等于MY_TAG1;
      如果相等,消息报消费;如果不相等,那么消息就会被忽略;


2、Producer.java的代码

[java] view plain copy
  1. import javax.jms.Connection;  
  2. import javax.jms.ConnectionFactory;  
  3. import javax.jms.DeliveryMode;  
  4. import javax.jms.Destination;  
  5. import javax.jms.JMSException;  
  6. import javax.jms.MapMessage;  
  7. import javax.jms.MessageProducer;  
  8. import javax.jms.Session;  
  9. import javax.jms.TextMessage;  
  10.   
  11. import org.apache.activemq.ActiveMQConnectionFactory;  
  12.   
  13. public class Producer {  
  14.   
  15.     // 单例模式  
  16.   
  17.     // 1、连接工厂  
  18.     private ConnectionFactory connectionFactory;  
  19.     // 2、连接对象  
  20.     private Connection connection;  
  21.     // 3、Session对象  
  22.     private Session session;  
  23.     // 4、生产者  
  24.     private MessageProducer messageProducer;  
  25.   
  26.     public Producer() {  
  27.   
  28.         try {  
  29.             this.connectionFactory = new ActiveMQConnectionFactory("zhangsan",  
  30.                     "123""tcp://localhost:61616");  
  31.             this.connection = connectionFactory.createConnection();  
  32.             this.connection.start();  
  33.             // 设置自动签收模式  
  34.             this.session = this.connection.createSession(false,  
  35.                     Session.AUTO_ACKNOWLEDGE);  
  36.             this.messageProducer = this.session.createProducer(null);  
  37.         } catch (JMSException e) {  
  38.             throw new RuntimeException(e);  
  39.         }  
  40.   
  41.     }  
  42.   
  43.     public Session getSession() {  
  44.         return this.session;  
  45.     }  
  46.   
  47.     public void send1(/* String QueueName, Message message */) {  
  48.         try {  
  49.   
  50.             Destination destination = this.session.createQueue("first");  
  51.             MapMessage msg1 = this.session.createMapMessage();  
  52.             msg1.setString("name""张三");  
  53.             msg1.setInt("age"20);  
  54.             // 设置用于消息过滤器的条件  
  55.             msg1.setStringProperty("name""张三");  
  56.             msg1.setIntProperty("age"20);  
  57.             msg1.setStringProperty("color""bule");  
  58.   
  59.             MapMessage msg2 = this.session.createMapMessage();  
  60.             msg2.setString("name""李四");  
  61.             msg2.setInt("age"25);  
  62.             // 设置用于消息过滤器的条件  
  63.             msg2.setStringProperty("name""李四");  
  64.             msg2.setIntProperty("age"25);  
  65.             msg2.setStringProperty("color""white");  
  66.   
  67.             MapMessage msg3 = this.session.createMapMessage();  
  68.             msg3.setString("name""张三");  
  69.             msg3.setInt("age"30);  
  70.             // 设置用于消息过滤器的条件  
  71.             msg3.setStringProperty("name""赵六");  
  72.             msg3.setIntProperty("age"30);  
  73.             msg3.setStringProperty("color""black");  
  74.   
  75.             // 发送消息  
  76.             this.messageProducer.send(destination, msg1,  
  77.                     DeliveryMode.NON_PERSISTENT, 41000 * 60 * 10);  
  78.             this.messageProducer.send(destination, msg2,  
  79.                     DeliveryMode.NON_PERSISTENT, 41000 * 60 * 10);  
  80.             this.messageProducer.send(destination, msg3,  
  81.                     DeliveryMode.NON_PERSISTENT, 41000 * 60 * 10);  
  82.   
  83.         } catch (JMSException e) {  
  84.             throw new RuntimeException(e);  
  85.         }  
  86.     }  
  87.   
  88.     public void send2() {  
  89.         try {  
  90.             Destination destination = this.session.createQueue("first");  
  91.             TextMessage message = this.session.createTextMessage("我是一个字符串");  
  92.             // 发送消息  
  93.             this.messageProducer.send(destination, message,  
  94.                     DeliveryMode.NON_PERSISTENT, 41000 * 60 * 10);  
  95.         } catch (JMSException e) {  
  96.             throw new RuntimeException(e);  
  97.         }  
  98.   
  99.     }  
  100.   
  101.     public static void main(String[] args) {  
  102.         Producer producer = new Producer();  
  103.         producer.send1();  
  104.   
  105.     }  
  106.   
  107. }  



3、Conmuser.java的代码

[java] view plain copy
  1. import javax.jms.Connection;  
  2. import javax.jms.ConnectionFactory;  
  3. import javax.jms.Destination;  
  4. import javax.jms.JMSException;  
  5. import javax.jms.MapMessage;  
  6. import javax.jms.Message;  
  7. import javax.jms.MessageConsumer;  
  8. import javax.jms.MessageListener;  
  9. import javax.jms.Session;  
  10. import javax.jms.TextMessage;  
  11.   
  12. import org.apache.activemq.ActiveMQConnectionFactory;  
  13.   
  14. public class Conmuser {  
  15.   
  16.     // 单例模式  
  17.   
  18.     // 1、连接工厂  
  19.     private ConnectionFactory connectionFactory;  
  20.     // 2、连接对象  
  21.     private Connection connection;  
  22.     // 3、Session对象  
  23.     private Session session;  
  24.     // 4、生产者  
  25.     private MessageConsumer messageConsumer;  
  26.     // 5、目的地址  
  27.     private Destination destination;  
  28.   
  29.     // 消息选择器  
  30.     public final String SELECTOR_1 = "age > 20";  
  31.   
  32.     public Conmuser() {  
  33.   
  34.         try {  
  35.             this.connectionFactory = new ActiveMQConnectionFactory("zhangsan",  
  36.                     "123""tcp://localhost:61616");  
  37.             this.connection = connectionFactory.createConnection();  
  38.             this.connection.start();  
  39.             // 设置自动签收模式  
  40.             this.session = this.connection.createSession(false,  
  41.                     Session.AUTO_ACKNOWLEDGE);  
  42.             this.destination = this.session.createQueue("first");  
  43.             // 在构造消费者的时候,指定了 消息选择器  
  44.             // 有选择性的消费消息  
  45.             this.messageConsumer = this.session.createConsumer(destination,  
  46.                     SELECTOR_1);  
  47.         } catch (JMSException e) {  
  48.             throw new RuntimeException(e);  
  49.         }  
  50.   
  51.     }  
  52.   
  53.     public Session getSession() {  
  54.         return this.session;  
  55.     }  
  56.   
  57.     // 用于监听消息队列的消息  
  58.     class MyLister implements MessageListener {  
  59.   
  60.         @Override  
  61.         public void onMessage(Message message) {  
  62.             try {  
  63.                 if (message instanceof TextMessage) {  
  64.   
  65.                 }  
  66.                 if (message instanceof MapMessage) {  
  67.                     MapMessage ret = (MapMessage) message;  
  68.                     System.out.println(ret.toString());  
  69.                     System.out.println(ret.getString("name"));  
  70.                     System.out.println(ret.getInt("age"));  
  71.                 }  
  72.             } catch (JMSException e) {  
  73.                 throw new RuntimeException(e);  
  74.             }  
  75.         }  
  76.   
  77.     }  
  78.   
  79.     // 用于异步监听消息  
  80.     public void receiver() {  
  81.         try {  
  82.             this.messageConsumer.setMessageListener(new MyLister());  
  83.         } catch (JMSException e) {  
  84.             throw new RuntimeException(e);  
  85.         }  
  86.     }  
  87.   
  88.     public static void main(String[] args) {  
  89.         Conmuser conmuser = new Conmuser();  
  90.         conmuser.receiver();  
  91.   
  92.     }  
  93.   
  94. }  





4、需要注意的地方

      4.1  注意消息过滤器的过滤条件的设置

[java] view plain copy
  1. // 设置用于消息过滤器的条件  
  2. msg3.setStringProperty("name""赵六");  
  3. msg3.setIntProperty("age"30);  
  4. msg3.setStringProperty("color""black");  
       

       4.2 消息过滤器的写法(类似于SQL语句的写法)

[java] view plain copy
  1. // 消息选择器  
  2. public final String SELECTOR_1 = "age > 20";  
  3. public final String SELECTOR_2 = " age > 20 and color='bule'"
相关文章
相关标签/搜索