ActiveMQ的消息过滤器

ゝ一世哀愁。 2022-07-15 04:10 107阅读 0赞

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的代码

import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MapMessage;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Producer {
    
    	// 单例模式
    
    	// 1、连接工厂
    	private ConnectionFactory connectionFactory;
    	// 2、连接对象
    	private Connection connection;
    	// 3、Session对象
    	private Session session;
    	// 4、生产者
    	private MessageProducer messageProducer;
    
    	public Producer() {
    
    		try {
    			this.connectionFactory = new ActiveMQConnectionFactory("zhangsan",
    					"123", "tcp://localhost:61616");
    			this.connection = connectionFactory.createConnection();
    			this.connection.start();
    			// 设置自动签收模式
    			this.session = this.connection.createSession(false,
    					Session.AUTO_ACKNOWLEDGE);
    			this.messageProducer = this.session.createProducer(null);
    		} catch (JMSException e) {
    			throw new RuntimeException(e);
    		}
    
    	}
    
    	public Session getSession() {
    		return this.session;
    	}
    
    	public void send1(/* String QueueName, Message message */) {
    		try {
    
    			Destination destination = this.session.createQueue("first");
    			MapMessage msg1 = this.session.createMapMessage();
    			msg1.setString("name", "张三");
    			msg1.setInt("age", 20);
    			// 设置用于消息过滤器的条件
    			msg1.setStringProperty("name", "张三");
    			msg1.setIntProperty("age", 20);
    			msg1.setStringProperty("color", "bule");
    
    			MapMessage msg2 = this.session.createMapMessage();
    			msg2.setString("name", "李四");
    			msg2.setInt("age", 25);
    			// 设置用于消息过滤器的条件
    			msg2.setStringProperty("name", "李四");
    			msg2.setIntProperty("age", 25);
    			msg2.setStringProperty("color", "white");
    
    			MapMessage msg3 = this.session.createMapMessage();
    			msg3.setString("name", "张三");
    			msg3.setInt("age", 30);
    			// 设置用于消息过滤器的条件
    			msg3.setStringProperty("name", "赵六");
    			msg3.setIntProperty("age", 30);
    			msg3.setStringProperty("color", "black");
    
    			// 发送消息
    			this.messageProducer.send(destination, msg1,
    					DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
    			this.messageProducer.send(destination, msg2,
    					DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
    			this.messageProducer.send(destination, msg3,
    					DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
    
    		} catch (JMSException e) {
    			throw new RuntimeException(e);
    		}
    	}
    
    	public void send2() {
    		try {
    			Destination destination = this.session.createQueue("first");
    			TextMessage message = this.session.createTextMessage("我是一个字符串");
    			// 发送消息
    			this.messageProducer.send(destination, message,
    					DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
    		} catch (JMSException e) {
    			throw new RuntimeException(e);
    		}
    
    	}
    
    	public static void main(String[] args) {
    		Producer producer = new Producer();
    		producer.send1();
    
    	}
    
    }

3、Conmuser.java的代码

import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MapMessage;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Conmuser {
    
    	// 单例模式
    
    	// 1、连接工厂
    	private ConnectionFactory connectionFactory;
    	// 2、连接对象
    	private Connection connection;
    	// 3、Session对象
    	private Session session;
    	// 4、生产者
    	private MessageConsumer messageConsumer;
    	// 5、目的地址
    	private Destination destination;
    
    	// 消息选择器
    	public final String SELECTOR_1 = "age > 20";
    
    	public Conmuser() {
    
    		try {
    			this.connectionFactory = new ActiveMQConnectionFactory("zhangsan",
    					"123", "tcp://localhost:61616");
    			this.connection = connectionFactory.createConnection();
    			this.connection.start();
    			// 设置自动签收模式
    			this.session = this.connection.createSession(false,
    					Session.AUTO_ACKNOWLEDGE);
    			this.destination = this.session.createQueue("first");
    			// 在构造消费者的时候，指定了 消息选择器
    			// 有选择性的消费消息
    			this.messageConsumer = this.session.createConsumer(destination,
    					SELECTOR_1);
    		} catch (JMSException e) {
    			throw new RuntimeException(e);
    		}
    
    	}
    
    	public Session getSession() {
    		return this.session;
    	}
    
    	// 用于监听消息队列的消息
    	class MyLister implements MessageListener {
    
    		@Override
    		public void onMessage(Message message) {
    			try {
    				if (message instanceof TextMessage) {
    
    				}
    				if (message instanceof MapMessage) {
    					MapMessage ret = (MapMessage) message;
    					System.out.println(ret.toString());
    					System.out.println(ret.getString("name"));
    					System.out.println(ret.getInt("age"));
    				}
    			} catch (JMSException e) {
    				throw new RuntimeException(e);
    			}
    		}
    
    	}
    
    	// 用于异步监听消息
    	public void receiver() {
    		try {
    			this.messageConsumer.setMessageListener(new MyLister());
    		} catch (JMSException e) {
    			throw new RuntimeException(e);
    		}
    	}
    
    	public static void main(String[] args) {
    		Conmuser conmuser = new Conmuser();
    		conmuser.receiver();
    
    	}
    
    }

4、需要注意的地方

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

// 设置用于消息过滤器的条件
    msg3.setStringProperty("name", "赵六");
    msg3.setIntProperty("age", 30);
    msg3.setStringProperty("color", "black");

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

// 消息选择器
    	public final String SELECTOR_1 = "age > 20";
    	public final String SELECTOR_2 = " age > 20 and color='bule'";