JMS学习四（ActiveMQ消息过滤）

Posted 2020-11-02 南望孤笑

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了JMS学习四（ActiveMQ消息过滤）相关的知识，希望对你有一定的参考价值。

消息的过期、消息的选择器和消息的优先级。

一、消息的过期

允许消息过期。默认情况下，消息永不会过期。如果消息在特定周期内失去意义，那么可以设置过期时间。
有两种方法设置消息的过期时间，时间单位为毫秒：
1．使用消息生产者的setTimeToLive 方法为所有的消息设置过期时间。
2．使用消息生产者的send 方法为每一条消息设置过期时间。
消息过期时间，send 方法中的 timeToLive 值加上发送时刻的 GMT 时间值。如果 timeToLive 值等于零，则 JMSExpiration 被设为零，表示该消息永不过期。

3、消息服务器接收到消息后，在指定的时间后，会从队列中移除指定的消息，超时被移除的消息不会发送给消费者。

4、使用消息生产者的setTimeToLive(long time ) 方法来给所有的消息设置过期时间：

// 消息生产者  
MessageProducer producer = null;  
producer = session.createProducer(queue);  
// 消息是否为持久性的，如果不设置默认是持久化的。  
producer.setDeliveryMode(DeliveryMode.PERSISTENT);  
 //消息过期设置  
producer.setTimeToLive(1000);

5、使用消息生产者的send()方法来设置消息的过期时间

//message发送的消息，deliveryMode是否持久化，priority优先级，timeToLive 消息过期时间  
//producer.send(message, deliveryMode, priority, timeToLive);  
producer.send(message, DeliveryMode.PERSISTENT, 4, 1000);

这里在插一段吧，上面设置消息过期的都是消息生产者这方的来设置的，也就是如果不满足条件则消息服务器会把消息从消息队列中删除，但是我们也可以在消息消费端来设置接受时间（仅限于同步接受）

Message message = consumer.receive(2);

就是在接受的时候添加等待时间（单位是毫秒）如果在指定的时间内获取不到消息则不会再等了。如果不设置等待时间则一直等待直到接收到消息或超时为止。

二、消息的选择器

不管是在消息发送端设置消息过期时间还是在接收端设置等待时间，都是对不满足的消息有过滤的作用，那消息选择器就是为过滤消息而生的下面来看看消息选择器：

ActiveMQ提供了一种机制，使用它，消息服务可根据消息选择器中的标准来执行消息过滤。生产者可在消息中放入应用程序特有的属性，而消费者可使用基于这些属性的选择标准来表明对消息是否感兴趣。这就简化了客户端的工作，并避免了向不需要这些消息的消费者传送消息的开销。然而，它也使得处理选择标准的消息服务增加了一些额外开销。消息选择器是用于MessageConsumer的过滤器，可以用来过滤传入消息的属性和消息头部分（但不过滤消息体），并确定是否将实际消费该消息。消息选择器是一些字符串，它们基于某种语法，而这种语法是SQL-92的子集。可以将消息选择器作为MessageConsumer 创建的一部分。

1、消息生产者：

package mqtest3;  
  
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("admin",  
                    "admin", "tcp://127.0.0.1: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("我是一个字符串");  
            message.setIntProperty("age", 25);  
            // 发送消息  
            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();  
        // producer.send2();  
  
    }  
}

2、消息消费者：

package mqtest3;  
  
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 > 25";  
    public final String SELECTOR_2 = " age > 20 and color=‘black‘";  
  
    public Conmuser() {  
        try {  
            this.connectionFactory = new ActiveMQConnectionFactory("admin",  
                    "admin", "tcp://127.0.0.1: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) {  
                    TextMessage ret = (TextMessage) message;  
                    System.out.println("results;" + ret.getText());  
                }  
                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();  
  
    }  
}

上面的demo是对MapMessage和TextMessage两种消息的过滤条件的设置和消费，过滤条件的设置使在消息的属性中设置，而消费消息的时候直接是在session创建MessageConsumer时传入的参数即过滤条件（过滤条件的写法和SQL的写法是很像的）

在写过滤条件的时候要注意设置的是什么类型的条件即： int 、string 如果是int 则加引号而如果是String则要加哦！！！

三、消息的优先级

通常，可以确保将单个会话向目标发送的所有消息按其发送顺序传送至消费者。然而，如果为这些消息分配了不同的优先级，消息传送系统将首先尝试传送优先级较高的消息。
有两种方法设置消息的优先级：
1．使用 setPriority 方法，这样所有的消息都采用此传送模式；
2．使用 send 方法为每一条消息设置传送模式；
消息优先级从 0-9 十个级别，0-4 是普通消息，5-9 是加急消息。如果不指定优先级，则默认为 4。JMS 不要求严格按照这十个优先级发送消息，但必须保证加急消息要先于普通消息到达。

以上是关于JMS学习四（ActiveMQ消息过滤）的主要内容，如果未能解决你的问题，请参考以下文章

WSO2 ESB 5.0.0 配置 JMS 传输（ActiveMQ）- 主题消息发布与订阅

JMS中间件ActiveMQ详解

JMS学习六（ActiveMQ消息传送模型）

学习ActiveMQ：JMS消息的事务管理