尚硅谷ActiveMQ学习笔记（2）-- Java编码实现ActiveMQ通讯

叁歲伎倆 2022-11-13 12:23 169阅读 0赞

### 一、IDEA创建Maven工程 ###

### 二、POM.xml ###

<?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>top.xf.activemq</groupId>
        <artifactId>activemq-demo</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <maven.compiler.source>1.8</maven.compiler.source>
            <maven.compiler.target>1.8</maven.compiler.target>
        </properties>
    
        <dependencies>
            
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
                <version>5.15.9</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.xbean</groupId>
                <artifactId>xbean-spring</artifactId>
                <version>3.16</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
                <version>1.7.25</version>
            </dependency>
            <dependency>
                <groupId>ch.qos.logback</groupId>
                <artifactId>logback-classic</artifactId>
                <version>1.2.3</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.16.18</version>
                <scope>provided</scope>
            </dependency>
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.12</version>
            </dependency>
        </dependencies>
    </project>

### 三、JMS编码总体架构 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70]  
之前JDBC的编码套路

通过JDBC操作数据库通用步骤：
    第1步：注册驱动(仅仅做一次)                     Class.forName(“com.mysqljdbc.Driver”)；
     
    第2步：建立连接(Connection)                       Connection conn =DriverManager.getConnection(url，user，password)；
     
    第3步：创建运行SQL的语句(Statement)      Statement st=connection.createStatement()；
     
    第4步：运行语句                                            ResutSet rs =stexecuteQuery(sql)；
     
    第5步：处理运行结果（ResultSet）             省略……
     
    第6步：释放资源                                            省略……

### 四、Destination队列和主题 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 1]

### 五、在点对点的消息传递域中，目的地被称为队列（queue） ###

#### 1、消费生产者 ####

##### ①、依赖 #####

<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 JmsProduce { 
    
        public static final String ACTIVEMQ_URL = "tcp://192.168.85.128:61616";
        public static final String QUEUE_NAME = "queue01";
    
        public static void main(String[] args) throws JMSException { 
    
            // 1.创建连接工厂,按照给定的url地址,采用默认的账号和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            // 2.通过连接工厂,获取连接Connection并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            // 3.创建会话session
            // 两个参数：第一个叫事务/第二个叫签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 4.创建目的地(具体是队列还是主题topic)
            Queue queue = session.createQueue(QUEUE_NAME);
            // 5.创建消息生产者
            MessageProducer messageProducer = session.createProducer(queue);
            // 6.通过使用MessageProducer生产3条消息到MQ的队列里面
            for (int i = 1; i <= 3; i++) { 
                // 7.创建消息，好比学生按照阳哥的要求写好的面试题消息
                TextMessage textMessage = session.createTextMessage("msg------" + i);//理解为一个字符串
                // 8.通过MessageProducer发送给mq
                messageProducer.send(textMessage);
            }
            // 9.关闭资源
            messageProducer.close();
            session.close();
            connection.close();
            System.out.println("消息发送到MQ完成");
        }
    }

##### ③、控制台 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 2]  
Number Of Pending Messages|等待消费的消息|这个是当前未出队列的数量。公式=总接收数-总出队列数

Number Of Consumers 消费者数量消费者端的消费者数量

Messages Enqueued |进队消息数|进入队列的总数量，包括出队列的。这个数量只增不减

Messages Dequeued |出队消息数|可以理解为是消费者消费掉的数量

总结：  
当有一个消息进入这个队列时，等待消费的消息是1，进入队列的消息是1。  
当消息消费后，等待消费的消息是0，进入队列的消息是1，出队列的消息是1  
再来一条消息时，等待消费的消息是1，进入队列的消息就是2

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 3]

##### ④、截图 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 5]

#### 2、消息消费者 ####

##### ①、代码1 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 6]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 7]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 8]

##### ②、代码2 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 9]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 10]

##### ③、全部代码 #####

import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    import java.io.IOException;
    
    /** * activemq消息的消费者 * * @author xiaofei */
    public class JmsConsumer { 
    
        public static final String ACTIVEMQ_URL = "tcp://192.168.85.128:61616";
        public static final String QUEUE_NAME = "queue01";
    
        public static void main(String[] args) throws JMSException, IOException { 
    
            System.out.println("2号消费者");
    
            // 1.创建连接工厂,按照给定的url地址,采用默认的账号和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            // 2.通过连接工厂,获取拦截Connection
            Connection connection = activeMQConnectionFactory.createConnection();
            // 3.启动
            connection.start();
            // 4.创建会话session
            // 两个参数：第一个叫事务/第二个叫签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 5.创建目的地(具体是队列还是主题topic)
            Queue queue = session.createQueue(QUEUE_NAME);
            // 6.创建消费者
            MessageConsumer consumer = session.createConsumer(queue);
    
    // 同步阻塞方式aeceive()
    // 订阅者或接受者调用MessageConsumer的receive()方法来接收消息,receive方法在能接收到消息之前(或超时之前)将一直阻塞
    
    // while (true) { 
    // // 生产消息是TextMessage消费需要强转成TextMessage
    // // receive里面可以设置时间限制
    // TextMessage receive = (TextMessage) consumer.receive(4000L);
    // if (null != receive) { 
    // System.out.println("接收到消息：" + receive.getText());
    // } else { 
    // break;
    // }
    // }
    // // 关闭资源
    // consumer.close();
    // session.close();
    // connection.close();
    
    // 同步监听方式来消费消息
            consumer.setMessageListener(new MessageListener() { 
                @Override
                public void onMessage(Message message) { 
                    if (null != message && message instanceof TextMessage) { 
                        TextMessage textMessage = (TextMessage) message;
                        try { 
                            System.out.println("接收到消息：" + textMessage.getText());
                        } catch (JMSException e) { 
                            e.printStackTrace();
                        }
                    }
                }
            });
            // 等待
            System.in.read();
            // 关闭资源
            consumer.close();
            session.close();
            connection.close();
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 11]

#### 3、编码总结 ####

##### ①、JMS开发的基本步骤 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 12]

##### ②、两种消费方式 #####

同步阻塞方式（receive()）
    订阅者或者接受者调用MessageConsumer的reveive()方法来接收消息,receive方法在能够接收到消息之前（或超时之前）将一直阻塞。
     
    异步非阻塞方式（监听器onMessage()）
    订阅者或者接收者调用MessageConsumer的setMessageListener注册一个消息监听器,
    当消息到达之后,系统自动调用监听器MessageListener的onMessage方法。

### 六、在发布订阅消息传递域中,目的地被称为主题（topic） ###

#### 1、发布主题生产者 ####

package top.xf.activemq.queue;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    import java.io.IOException;
    
    /** * 订阅消息的推送 * @author xiaofei */
    public class JmsProduce_Topic { 
    
        public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";
    
        public static final String TOPIC_NAME = "topic-xiaofei";
    
        public static void main(String[] args) throws JMSException, IOException { 
    
            System.err.println("2号消费者");
    
            // 1.创建连接工厂,按照给定的url地址,采用默认的账号和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            // 2.通过连接工厂,获取拦截Connection
            Connection connection = activeMQConnectionFactory.createConnection();
            // 3.启动
            connection.start();
            // 4.创建会话session
            // 两个参数：事务 / 签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 5.创建目的地(具体是队列还是主题topic)
            Topic topic = (Topic) session.createTopic(TOPIC_NAME);
            // 6.创建消息生产者
            MessageProducer producer = session.createProducer(topic);
            // 7.通过使用MessageProducer生产3条消息到MQ的队列里面
            for (int i = 1; i <= 6; i++) { 
                // 8.创建消息
                TextMessage textMessage = session.createTextMessage("msg------" + i);
                // 9.通过MessageProducer发送给mq
                producer.send(textMessage);
            }
            // 10.关闭资源
            producer.close();
            session.close();
            connection.close();
            System.err.println("TOPIC_NAME消息发送到MQ完成");
        }
    }

#### 2、订阅主题消费者 ####

package top.xf.activemq.queue;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    import java.io.IOException;
    
    /** * 订阅消息的接收 * * @author xiaofei */
    public class JmsConsumer_Topic { 
    
        public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";
    
        public static final String TOPIC_NAME = "topic-xiaofei";
    
        public static void main(String[] args) throws JMSException, IOException { 
    
            // 1.创建连接工厂,按照给定的url地址,采用默认的账号和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            // 2.通过连接工厂,获取拦截Connection
            Connection connection = activeMQConnectionFactory.createConnection();
            // 3.启动
            connection.start();
            // 4.创建会话session
            // 两个参数：事务 / 签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 5.创建目的地(具体是队列还是主题topic)
            Topic topic = (Topic) session.createTopic(TOPIC_NAME);
            // 6.创建消息生产者
            MessageConsumer consumer = session.createConsumer(topic);
            // 使用lambda表达式
            consumer.setMessageListener((message) -> { 
                if (null != message && message instanceof TextMessage) { 
                    TextMessage textMessage = (TextMessage) message;
                    try { 
                        System.err.println("接收到消息：" + textMessage.getText());
                    } catch (JMSException e) { 
                        e.printStackTrace();
                    }
                }
            });
            // 等待
            System.in.read();
            // 关闭资源
            consumer.close();
            session.close();
            connection.close();
        }
    
    }

#### 3、注意事项 ####

先启动订阅在启动生产,不然发送的消息是废消息

### 七、总结 ###

两大模式比较

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 13]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70]: /images/20221022/b69198b566ac41969def90de2e7ab9a7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 1]: /images/20221022/d65d3f61d16d4e95b8affae9a1ad9e83.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210328103717747.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20210329130107623.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20210328103850679.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20210328103857540.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/2021032912524984.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/2021032912525732.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20210329125451124.jpeg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20210329125645150.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 10]: https://img-blog.csdnimg.cn/20210329125655555.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 11]: https://img-blog.csdnimg.cn/20210329125739910.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 12]: https://img-blog.csdnimg.cn/20210329125805302.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM_size_16_color_FFFFFF_t_70 13]: https://img-blog.csdnimg.cn/20210330225452488.jpeg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V4b2R1czM=,size_16,color_FFFFFF,t_70