ActiveMQ(2)-一些配置

朱雀 2022-06-05 00:07 134阅读 0赞

上一篇初探，设置了一些固定的设置。本篇修改一些参数设置，查看一下效果。

# 1.事务 #

package com.mq.activemq_01.shiwu;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Sender { 
    
        public static void main(String[] args) throws JMSException {
    
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                                                                                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                                                                                "tcp://localhost:61616");
    
            Connection connection = connectionFactory.createConnection();
            connection.start();
    
            //3.通过Connection创建Session会话，参数1表示是否启动事务，参数2表示签收模式封装在Session中
            //修改是否启用事务为是，这时候提交的消息需要commit，MQ才可以真实的接收到
            Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    
            Destination destination = session.createQueue("queue2");
            MessageProducer messageProducer = session.createProducer(destination);
            messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    
            for (int i = 1; i <= 10; i++) {
                TextMessage textMessage = session.createTextMessage();
                textMessage.setText("我是消息内容，id="+i);
                messageProducer.send(textMessage);
                System.out.println("生产者：生产了"+textMessage.getText());
            }
    
            //如果创建的Session是需要使用事务提交消息（注释3），必须commit()
            session.commit();
    
            if(connection != null){
                connection.close();
            }
    
        }
    }

# 2.消息签收 #

第一篇的时候设置的是自动签收，理论上讲，不可能让消息自动签收的，需要消费者签收才算这条消息传输完成。

package com.mq.activemq_01.qianshou;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Sender { 
    
    
        public static void main(String[] args) throws JMSException {
    
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,                                                                 ActiveMQConnectionFactory.DEFAULT_PASSWORD,         "tcp://localhost:61616");
    
            Connection connection = connectionFactory.createConnection();
            connection.start();
    
            //3.通过Connection创建Session会话，参数1表示是否启动事务，参数2表示签收模式封装在Session中
            //修改是否启用事务为是，这时候提交的消息需要commit，MQ才可以真实的接收到
            //修改签收方式为客户端签收
            Session session = connection.createSession(Boolean.TRUE,Session.CLIENT_ACKNOWLEDGE);
    
            Destination destination = session.createQueue("queue2");
    
            MessageProducer messageProducer = session.createProducer(destination);
            messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    
            for (int i = 1; i <= 10; i++) {
                TextMessage textMessage = session.createTextMessage();
                textMessage.setText("我是消息内容，id="+i);
                messageProducer.send(textMessage);  
                System.out.println("生产者：生产了"+textMessage.getText());
            }
            //如果创建的Session是需要使用事务提交消息，必须commit()
            session.commit();
            if(connection != null){
                connection.close();
            }
        }
    }

consumer

package com.mq.activemq_01.qianshou;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Receiver { 
    
    
        public static void main(String[] args) throws JMSException {
    
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
    ActiveMQConnectionFactory.DEFAULT_PASSWORD,
    "tcp://localhost:61616");
    
            Connection connection = connectionFactory.createConnection();
            connection.start();
    
            //这里设置的签收方式为CLIENT_ACKNOWLEDGE客户端签收，如果客户端不签收，那么服务端就一直认为该消息未被消费。
            //签收需要调用message.acknowledge()方法，在while循环中
            Session session = connection.createSession(Boolean.FALSE, Session.CLIENT_ACKNOWLEDGE);
    
            Destination destination = session.createQueue("queue2");
    
            MessageConsumer messageConsumer = session.createConsumer(destination);
    
            while (true) {
                TextMessage receive = (TextMessage)messageConsumer.receive();
                //接收后，设置签收，MQ才会认为消息已经签收，否则一直停留在服务端
                receive.acknowledge();
                if(receive == null){
                    break;
                }
                System.out.println("消费者：收到消息"+receive.getText());
            }
    
            //7.释放连接
            if(connection != null){
                connection.close();
            }
        }
    }

# 3.消息过滤 #

消费者通过一定的条件，收取指定条件的消息。但是通过setInt(),setString()等是无法过滤的，过滤条件为设置的setIntProperty(),setStringProperty()属性值。如下面的实例，过滤条件1,2是无法过滤的，过滤条件3才可以。

package com.mq.activemq_01.selector;
    
    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;
    
    public class Producer { 
    
        private ConnectionFactory connectionFactory;
        private Connection connection;
        private Session session;
        private MessageProducer messageProducer;
    
        public Producer(){
            try {
    
                this.connectionFactory = new ActiveMQConnectionFactory( ActiveMQConnectionFactory.DEFAULT_USER,
                                                                        ActiveMQConnectionFactory.DEFAULT_PASSWORD, 
                                                                        "tcp://localhost:61616");
                this.connection = this.connectionFactory.createConnection();
                this.connection.start();
                this.session = this.connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
                this.messageProducer = this.session.createProducer(null);
    
            } catch (Exception e) {
                // TODO: handle exception
            }
        }
    
        public void send1(){
            try {
                Destination destination = this.session.createQueue("one");
    
                MapMessage msg1 = this.session.createMapMessage();
                msg1.setString("loginName", "zs");
                msg1.setString("name", "zhangsan");
                msg1.setString("password", "123456");
                msg1.setInt("gender", 1);
                msg1.setInt("age", 9);
                msg1.setStringProperty("receiver", "A");
    
                MapMessage msg2 = this.session.createMapMessage();
                msg2.setString("loginName", "zs");
                msg2.setString("name", "zhangsan");
                msg2.setString("password", "123456");
                msg2.setInt("gender", 1);
                msg2.setInt("age", 19);
                msg2.setStringProperty("receiver", "B");
    
                MapMessage msg3 = this.session.createMapMessage();
                msg3.setString("loginName", "zs");
                msg3.setString("name", "zhangsan");
                msg3.setString("password", "123456");
                msg3.setInt("gender", 1);
                msg3.setInt("age", 9);
                msg3.setStringProperty("receiver", "A");
    
                MapMessage msg4 = this.session.createMapMessage();
                msg4.setString("loginName", "ls");
                msg4.setString("name", "lisi");
                msg4.setString("password", "123456");
                msg4.setInt("age", 29);
                msg4.setStringProperty("receiver", "A");
                //设置消息的超时时间为2分钟，超过两分钟没有消费者消费，消息自动失效
                this.messageProducer.send(destination, msg1, DeliveryMode.NON_PERSISTENT, 2, 1000*60*2);
                this.messageProducer.send(destination, msg2, DeliveryMode.NON_PERSISTENT, 3, 1000*60*2);
                this.messageProducer.send(destination, msg3, DeliveryMode.NON_PERSISTENT, 7, 1000*60*2);
                this.messageProducer.send(destination, msg4, DeliveryMode.NON_PERSISTENT, 8, 1000*60*2);
    
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) {
            Producer p = new Producer();
            p.send1();
        }
    }

consumer

package com.mq.activemq_01.selector;
    
    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 Consumer { 
    
        //selector真实过滤的并不是setString值，而是设置的setStringProperty属性
        public final String SELECTOR_1 = "name='zhangsan'";
        public final String SELECTOR_2 = "name='zhangsan' and age > 10";
        public final String SELECTOR_3 = "receiver = 'B'";
    
        private ConnectionFactory connectionFactory;
        private Connection connection;
        private Session session;
        private MessageConsumer messageConsumer;
        private Destination destination;
    
    
        public Consumer(){
            try {
                this.connectionFactory = new ActiveMQConnectionFactory( ActiveMQConnectionFactory.DEFAULT_USER,
                                                                        ActiveMQConnectionFactory.DEFAULT_PASSWORD, 
                                                                        "tcp://localhost:61616");
                this.connection = this.connectionFactory.createConnection();
                this.connection.start();
                this.session = this.connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
                this.destination = this.session.createQueue("one");
    // this.messageConsumer = this.session.createConsumer(destination, SELECTOR_1);
                this.messageConsumer = this.session.createConsumer(destination, SELECTOR_3);
    // this.messageConsumer = this.session.createConsumer(destination);
    
            } catch (Exception e) {
                // TODO: handle exception
            }
        }
    
        public void receiver(){
            try {
                this.messageConsumer.setMessageListener(new Listener());
            } catch (Exception e) {
                // TODO: handle exception
            }
        }
        //提供监听类，放弃之前的while循环
        class Listener implements MessageListener{
    
            public void onMessage(Message message) {
                try {
                    if(message instanceof MapMessage){
                        MapMessage mm = (MapMessage) message;
    
                        System.out.println(mm.toString());
    
                        String name = mm.getString("name");
                        String loginName = mm.getString("loginName");
                        String password = mm.getString("password");
                        int age = mm.getInt("age");
    
                        System.out.println(name);
                        System.out.println(loginName);
                        System.out.println(password);
                        System.out.println(age);
    
                    }
                    if(message instanceof TextMessage){
                        //
                    }
                } catch (JMSException e) {
                    e.printStackTrace();
                }
    
            }
    
        }
    
        public static void main(String[] args) {
            Consumer consumer = new Consumer();
            consumer.receiver();
        }
    }

# 4.独立消费 #

消息可以设置为普通消息和加急消息，这里做一下activemq提供的消息消费优先级配置。

package com.mq.activemq_01.dulixiaofei;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Sender { 
        public static void main(String[] args) throws JMSException {
    
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,                                                                         ActiveMQConnectionFactory.DEFAULT_PASSWORD,
    "tcp://localhost:61616");
    
            Connection connection = connectionFactory.createConnection();
            connection.start();
    
            Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    
            Destination destination = session.createQueue("queue2");
    
            //5.创建消息发送或者接收对象
            MessageProducer messageProducer = session.createProducer(null);
                messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    
            //7.通过session创建message，JMS定义了五种正文格式
            for (int i = 1; i <= 10; i++) {
    
                TextMessage textMessage = session.createTextMessage();
    
                textMessage.setText("我是消息内容，id="+i);
    
    // messageProducer.send(textMessage);
                //destination消息发送目的地
                //textMessage消息内容
                //DeliveryMode.NON_PERSISTENT消息非持久化
                //i-1表示消息的优先级（0-9,0-4表示普通，5-9表示加急）
                //1000*60*2消息的有效时间，这里设置为2分钟
    
                //activemq.xml配置
                //
                //<policyEntry queue="queue2" prioritizedMessages="true"/>
                messageProducer.send(destination, textMessage, DeliveryMode.NON_PERSISTENT,i-1, 1000*60*2);
    
                System.out.println("生产者：生产了"+textMessage.getText());
            }
    
            session.commit();
    
            if(connection != null){
                connection.close();
            }
    
        }
    }

代码中虽然分别设置了消费优先级从0-9，但是实际消费的时候发现并没有按照优先级进行消费。这里需要做一些配置。  
打开apachemq.xml文件，做如下配置

![这里写图片描述][SouthEast]

执行结果：

> 消费者：收到消息我是消息内容，id=10  
> 消费者：收到消息我是消息内容，id=9  
> 消费者：收到消息我是消息内容，id=8  
> 消费者：收到消息我是消息内容，id=7  
> 消费者：收到消息我是消息内容，id=6  
> 消费者：收到消息我是消息内容，id=5  
> 消费者：收到消息我是消息内容，id=4  
> 消费者：收到消息我是消息内容，id=3  
> 消费者：收到消息我是消息内容，id=2  
> 消费者：收到消息我是消息内容，id=1

# 5.数据持久化 #

数据持久化放在下一节ActiveMQ(3)-ActiveMQ+MySQL持久化。

[SouthEast]: /images/20220605/1043d377fba94dba98b52d72146dacfd.png