本文共 5592 字,大约阅读时间需要 18 分钟。
MessageConsumer的消息选择器
MessageConsumer是一个由Session创建的对象,用来从Destination接收消息。
看一下Session创建MessageConsumer的构造方法有哪些:
public MessageConsumer createConsumer(Destination destination) throws JMSException; public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException; public MessageConsumer createConsumer(Destination destination, String messageSelector, boolean noLocal) throws JMSException;public MessageConsumer createConsumer(Destination destination, tring messageSelector, boolean noLocal, MessageListener messageListener) throws JMSException ;public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException;public TopicSubscriber createDurableSubscriber(Topic topic, String name, String messageSelector, boolean noLocal) throws JMSException;
其中
messageSelector为消息选择器;(采用SQL92的语法)
public final string SELECTOR = "color = 'blue'";该选择器检查了传入消息的color属性,并检查这个属性的值是否等于blue,如果相等则消息被消费,否则消息被忽略。
noLocal标识默认为false,当设置为true时,限制消费者只能接收和自己相同的连接(Connection)所发的消息,此标识只适用于主题topic,不适用于队列queue。
name标识订阅主题所对应的订阅名称,持久订阅时需要设置此参数;
我们接下来看下消息选择器是怎么使用的?
先看下生产者代码:
package jeff.mq.p2p;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 org.apache.activemq.ActiveMQConnectionFactory;/** * @author jeffSheng * 2018年7月3日 */public class Producer { //1连接工厂 private ConnectionFactory connectionFactory; //2连接对象 private Connection connection; //3Session对象 private Session session; //4生产者 private MessageProducer messageProducer; public Producer(){ try { this.connectionFactory = new ActiveMQConnectionFactory( "jeff", "123456", "tcp://localhost:61616"); this.connection = connectionFactory.createConnection(); this.connection.start(); this.session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); this.messageProducer = session.createProducer(null); } catch (JMSException e) { e.printStackTrace(); } } public Session getSession(){ return this.session; } public static void main(String[] args) throws Exception { Producer producer = new Producer(); producer.sender1(); } public void sender1() throws Exception{ Destination destination = this.session.createQueue("first"); MapMessage msg1 = this.session.createMapMessage(); msg1.setString("name", "张三"); msg1.setString("age", "23"); msg1.setStringProperty("color", "blue"); msg1.setIntProperty("sal", 2200); int id = 1; msg1.setInt("id", id); String receiver = id % 2 == 0 ? "A" : "B"; msg1.setStringProperty("receiver", receiver); MapMessage msg2 = this.session.createMapMessage(); msg2.setString("name", "李四"); msg2.setString("age", "26"); msg2.setStringProperty("color", "red"); msg2.setIntProperty("sal", 1300); id = 2; msg1.setInt("id", id); receiver = id % 2 == 0 ? "A" : "B"; msg2.setStringProperty("receiver", receiver); MapMessage msg3 = this.session.createMapMessage(); msg3.setString("name", "王五"); msg3.setString("age", "28"); msg3.setStringProperty("color", "green"); msg3.setIntProperty("sal", 1500); id = 3; msg3.setInt("id", id); receiver = id % 2 == 0 ? "A" : "B"; msg3.setStringProperty("receiver", receiver); MapMessage msg4 = this.session.createMapMessage(); msg4.setString("name", "赵六"); msg4.setString("age", "30"); msg4.setStringProperty("color", "blue"); msg4.setIntProperty("sal", 1800); id = 4; msg4.setInt("id", id); receiver = id % 2 == 0 ? "A" : "B"; msg4.setStringProperty("receiver", receiver); this.messageProducer.send(destination,msg1,DeliveryMode.NON_PERSISTENT,2,1000*60*10); this.messageProducer.send(destination,msg2,DeliveryMode.NON_PERSISTENT,3,1000*60*10); this.messageProducer.send(destination,msg3,DeliveryMode.NON_PERSISTENT,6,1000*60*10); this.messageProducer.send(destination,msg4,DeliveryMode.NON_PERSISTENT,9,1000*60*10); if(connection!=null){ connection.close(); } }}
创建了四个消息msg1,msg2,msg3,msg4,这四个消息的类型是MapMessage,存放的是id,name,color,age,sal,receiver,这几个属性信息,生产者将消息存入first队列中。
运行:
接下来我们看怎么使用消息选择器去有条件的获取消息,这个例子中我们使用的是OnMessage这种异步的监听消息方式,不再使用receive这种轮询方式。
Consumer代码:
我们定义了四个选择器:
public final String SELECTOR_0 = "age > 25"; public final String SELECTOR_1 = "color = 'blue'"; public final String SELECTOR_2 = "color = 'blue' and sal > 2000"; public final String SELECTOR_3 = "receiver = 'A'";
我们先将第一个选择器SELECTOR_0这个条件age>25放进去:
this.messageConsumer = session.createConsumer(this.destination,SELECTOR_0);
运行打印:理论上李四、王五、赵六都可以被消费,验证下,结果却没有任何消息被消费:
这是因为:虽然李四、王五、赵六满足age>25,但是age存放的语法是:
msg1.setString("age", "23");
这种对于选择器是无效的,只有类似于以下这种指定了数据类型的才是可以的:
msg1.setStringProperty("color", "blue"); msg1.setIntProperty("sal", 2200);
那么,我们试试SELECTOR_1?
public final String SELECTOR_1 = "color = 'blue'";
理论上张三和赵六可以被打印出来:
果然!所以队列中其实还有李四和王五没有被消费!其实我们也可以指定满足李四和王五的条件去选择消费,这里就不试验了!
mysql消息持久化
activemq默认的持久化策略是kahadb:
在对性能要求不高的情况下,我们可以采用mysql数据库来存储activemq的消息:
另外需要在broker节点的外定义id为mysql-ds的bean,如下:
接下来,我们在本地数据库中新建立一个hello_mq的数据库,然后记得将mysql的jdbc驱动,加入activemq安装文件的lib目录下:
需要注意,生产者发送数据的时候要【持久化】;
重新启动activemq:
很简单!
转载地址:https://jeffsheng.blog.csdn.net/article/details/80947308 如侵犯您的版权,请留言回复原文章的地址,我们会给您删除此文章,给您带来不便请您谅解!