Java消息中间件学习笔记四 -- ActiveMQ的使用,【队列模式】

╰半夏微凉° 2022-06-06 06:51 149阅读 0赞

### JMS代码演练 ###

#### 使用JMS接口规范连接ActiveMQ ####

*  创建生产者
 *  创建发布者
 *  创建消费者
 *  创建订阅者

首先回顾一下JMS编码结构之间的关系

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

#### 使用idea创建maven项目 ####

##### 引入依赖： #####

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.winter.jms</groupId>
        <artifactId>jms-test</artifactId>
        <version>1.0-SNAPSHOT</version>
    
    
        <dependencies>
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
                <version>5.9.0</version>
            </dependency>
        </dependencies>
    </project>

##### 查看项目结构： #####

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

##### 创建生产者： #####

package com.winter.jms.queue;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /** * * Created by Winterchen on 2017/11/1. */
    public class AppProducer { 
    
        private static final String url = "tcp://127.0.0.1:61616";
        private static final String queueName = "queue-test";
    
        public static void main(String[] args) throws JMSException {
    
            //1. 创建ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    
            //2. 创建Connection
            Connection connection = connectionFactory.createConnection();
    
            //3. 启动连接
            connection.start();
    
            //4. 创建会话
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            //5. 创建一个目标
            Destination destination = session.createQueue(queueName);
    
            //6. 创建一个生产者
            MessageProducer producer = session.createProducer(destination);
    
            for (int i = 0; i < 100; i++) {
                //7. 创建消息
                TextMessage textMessage = session.createTextMessage("test"+i);
                //8. 发布消息
                producer.send(textMessage);
                System.out.println("发送消息"+textMessage.getText());
    
            }
    
            //9. 关闭连接
             connection.close();
    
    
        }
    }

运行查看：

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

在消息队列中已经拥有了100条消息了，此时我们成功的在ActiveMQ中发布了消息。

##### 创建消费者： #####

**注意：其中很多代码都是复制过来的，在实际开发中不可能产生这样的代码，这样是为了使得逻辑清晰而为之**

package com.winter.jms.queue;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /** * * Created by Winterchen on 2017/11/1. */
    public class AppConsumer { 
    
        private static final String url = "tcp://127.0.0.1:61616";
        private static final String queueName = "queue-test";
    
        public static void main(String[] args) throws JMSException {
    
            //1. 创建ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
    
            //2. 创建Connection
            Connection connection = connectionFactory.createConnection();
    
            //3. 启动连接
            connection.start();
    
            //4. 创建会话
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            //5. 创建一个目标
            Destination destination = session.createQueue(queueName);
    
           //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) {
    
                     }
                 }
             });
    
            //9. 关闭连接
            //connection.close();
    
    
        }
    }

**消费者的连接不允许关闭的，因为消息的接收是异步的，会导致消息不能被消费。**  
启动运行，可以看到：

接收消息  = [test0]
    接收消息  = [test1]
    接收消息  = [test2]
    接收消息  = [test3]
    接收消息  = [test4]
    接收消息  = [test5]
    接收消息  = [test6]
    接收消息  = [test7]
    接收消息  = [test8]
    接收消息  = [test9]
    接收消息  = [test10]
    接收消息  = [test11]
    接收消息  = [test12]
    接收消息  = [test13]
    . . .

如果我们现在往ActiveMQ中再发布100条消息，然后再启动一个消费者，看看消息的消费情况：  
消费者1：

接收消息  = [test0]
    接收消息  = [test2]
    接收消息  = [test4]
    接收消息  = [test6]
    接收消息  = [test8]
    接收消息  = [test10]
    接收消息  = [test12]
    接收消息  = [test14]
    接收消息  = [test16]
    接收消息  = [test18]
    接收消息  = [test20]
    接收消息  = [test22]
    接收消息  = [test24]
    接收消息  = [test26]
    接收消息  = [test28]
    . . .

消费者2：

接收消息  = [test1]
    接收消息  = [test3]
    接收消息  = [test5]
    接收消息  = [test7]
    接收消息  = [test9]
    接收消息  = [test11]
    接收消息  = [test13]
    接收消息  = [test15]
    接收消息  = [test17]
    接收消息  = [test19]
    接收消息  = [test21]
    接收消息  = [test23]
    接收消息  = [test25]
    接收消息  = [test27]
    . . .

可以看出，两个消费者是平均消费队列中的所有消息。

[SouthEast]: /images/20220606/65e368e91bb94511aa4df0c814551e43.png
[SouthEast 1]: /images/20220606/7233ad1685d846769a189af71853c9d0.png
[SouthEast 2]: /images/20220606/5279b027b32147cda604a3f613509079.png