Java ActiveMQ 队列模式

﹏ヽ暗。殇╰゛Y 2022-06-07 10:50 183阅读 0赞

本文为在慕课网学习的笔记和个人体会，视频地址：http://www.imooc.com/learn/856

队列模式中，一条消息只能被一个消费者消费。![Center][]

创建好工程后，在 pom.xml 文件中引用 activemq 的相关包

<dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
                <version>5.9.0</version>
            </dependency>

创建一个生产者，用于向目的地中发送消息：

import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * 消息提供者，向消息中间件发送消息
     */
    public class AppProducer {
    
        //activemq服务器地址 61616是activemq默认的端口，记得在服务器中开启该端口
        private final static String URL = "tcp://192.168.202.129:61616";
        //创建队列名字
        private final static String QUEUE_NAME = "queue-test";
    
        public static void main(String[] args) throws JMSException {
            //1.创建activemq 连接工厂 ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
    
            //2.连接实例
            Connection connection = connectionFactory.createConnection();
    
            //3.启动连接
            connection.start();
    
            //4.创建会话 session
            //第一个参数意思：是否在事物中进行处理 ，第二个参数是设置应答模式
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            //5.创建发送消息的目的地
            Destination destination = session.createQueue(QUEUE_NAME);
    
            //6.创建生产者
            MessageProducer producer = session.createProducer(destination);
    
            //7.使用生产者向目的地发送消息
    
            for (int i = 0 ;i < 100 ;i++){
                //8.创建消息
                TextMessage textMessage = session.createTextMessage("我是第  "+i+"  条消息！");
    
                //9.使用生产者将消息发送至目的地
                //第一个参数是目的地，第二个参数是 要发送的信息
                producer.send(destination, textMessage);
    
                System.out.println("成功发送消息： "+textMessage.getText());
            }
    
            connection.close();
        }
    }

创建一个消息的消费者，用于从目的地获取消息：

import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * 队列消费者，用于接收队列生产者发送的消息
     */
    public class AppConsumer {
    
        //activemq服务器地址 61616是activemq默认的端口，记得在服务器中开启该端口
        private final static String URL = "tcp://192.168.202.129:61616";
        //创建队列名字 
        private final static String QUEUE_NAME = "queue-test";
    
        public static void main(String[] args) throws JMSException {
            //1.创建activemq 连接工厂 ConnectionFactory 
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
            //2.连接实例 
            Connection connection = connectionFactory.createConnection();
            //3.启动连接 
            connection.start();
            //4.创建会话 session 
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //第一个参数意思：是否在事物中进行处理 ，第二个参数是设置应答模式 
            //5.创建发送消息的目的地 
            Destination destination = session.createQueue(QUEUE_NAME);
            //6.创建消费者 
            MessageConsumer consumer = session.createConsumer(destination);
            // 7.创建一个监听器 
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        System.out.println("成功接收消息 " + textMessage.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

[Center]: /images/20220607/2f66f5eb12c44b3ba7d3178d3397f771.png