ActiveMQ专题6 ——ActiveMQ的Broker

小鱼儿 2022-11-05 12:56 179阅读 0赞

## 是什么？ ##

相当于一个内嵌式ActiveMQ服务器实例。其实就是实现了用代码的形式启动ActiveMQ将MQ嵌入到Java代码中，以便随时用随时启动，在用的时候再去启动这样能节省了资源，也保证了可用性。

用ActiveMQ Broker作为独立的消息服务器来构建Java应用。ActiveMQ也支持在vm中通信基于嵌入的broker，能够无缝的集成其他java应用。

换言之，类似SpringBoot内嵌了一个Tomcat服务器。

## 怎么用？ ##

POM依赖如下：

<dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
                <version>5.15.11</version>
            </dependency>
            
            <dependency>
                <groupId>org.apache.xbean</groupId>
                <artifactId>xbean-spring</artifactId>
                <version>4.15</version>
            </dependency>
            
            <dependency>
                <groupId>com.fasterxml.jackson.core</groupId>
                <artifactId>jackson-databind</artifactId>
                <version>2.10.1</version>
            </dependency>

JAVA代码如下：

package com.huazai.activemq.broker;
    
    import org.apache.activemq.broker.BrokerService;
    
    /**
     * @author pyh
     * @date 2021/1/6 23:01
     */
    public class EmbedBroker {
        
        public static void main(String[] args) throws Exception {
        
            // 创建broker服务实例
            BrokerService brokerService = new BrokerService();
            brokerService.setPopulateJMSXUserID(true);
            // 绑定地址，此处不是再使用linux服务器上的ip地址
            brokerService.addConnector("tcp://127.0.0.1:61616");
            // 启动服务
            brokerService.start();
        }
    }

启动main函数，再命令行上输入jps -l出现你编写的类名，则表示broker服务启动成功。  
![在这里插入图片描述][2021030617505744.png]

## 测试 ##

在连接linux服务器上ActiveMQ的基础上，把连接到linux服务器的ip地址改成本机地址即可。参考：[Java编码实现ActiveMQ通讯（Topic）][Java_ActiveMQ_Topic]

消息提供者代码如下

package com.huazai.activemq.broker;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * @author pyh
     * @date 2020/12/23 23:21
     */
    public class JMSProducer {
        
        // ActiveMQ服务地址
        public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";
        // 消息队列名称
        public static final String QUEUE_NAME = "broker-queue";
    
        public static void main(String[] args) throws Exception {
        
            // 1.创建给定ActiveMQ服务连接工厂，使用默认的用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            // 2.通过连接工厂，创建连接对象并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            // 3.创建会话，第一个参数为是否开启事务，第二个参数为签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 4.创建目的地（队列或者主题）
            Queue queue = session.createQueue(QUEUE_NAME);
            // 可以用父接口Destination接受
            // Destination queue = session.createQueue(QUEUE_NAME);
            // 5.创建消息的生产者
            MessageProducer producer = session.createProducer(queue);
            // 6.通过消息生产者生产6条消息发送MQ队列
            for (int i = 0; i < 3; i++) {
        
                // 7.创建消息
                TextMessage textMessage = session.createTextMessage("msg" + i + ":hello world");
                // 8.将消息发送到MQ
                producer.send(textMessage);
            }
            // 9.关闭资源
            producer.close();
            session.close();
            connection.close();
            System.out.println("finish");
        }
    
    }

消息消费者代码如下：

package com.huazai.activemq.broker;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * @author pyh
     * @date 2020/12/24 22:09
     */
    public class JMSConsumer {
        
        // ActiveMQ服务地址
        public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";
        // 消息队列名称，取消息必须和存消息的队列名称一致
        public static final String QUEUE_NAME = "broker-queue";
    
        public static void main(String[] args) throws Exception {
        
            // 1.创建给定ActiveMQ服务连接工厂，使用默认的用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            // 2.通过连接工厂，创建连接对象并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            // 3.创建会话，第一个参数为是否开启事务，第二个参数为签收
            Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
            // 4.创建目的地（队列或者主题）
            Queue queue = session.createQueue(QUEUE_NAME);
            // 5.创建消费者
            MessageConsumer consumer = session.createConsumer(queue);
            while (true) {
        
                // 接受消息根据生产者发送消息类型强类型转换
                TextMessage message = (TextMessage) consumer.receive();
                if (message != null) {
        
                    String text = message.getText();
                    System.out.println(text);
                    message.acknowledge();
    //                session.commit();
                } else {
        
                    break;
                }
            }
            consumer.close();
            session.close();
            connection.close();
        }
    }

先启动消息提供者，把消息发送到broker，再启动消息消费者，消费者能从队列中成功获取消息，则表示测试成功。  
![在这里插入图片描述][20210306175415171.png]

## 连更文章 ##

[ActiveMQ专题0 —— 前言说明][ActiveMQ_0 _]  
[ActiveMQ专题1 —— 入门概述][ActiveMQ_1 _]  
[ActiveMQ专题2 —— ActiveMQ下载和安装（Linux版）][ActiveMQ_2 _ ActiveMQ_Linux]  
[ActiveMQ专题3 ——Java编码实现ActiveMQ通讯（Queue）][ActiveMQ_3 _Java_ActiveMQ_Queue]  
[ActiveMQ专题4 ——ActiveMQ专题4 ——Java编码实现ActiveMQ通讯（Topic）][Java_ActiveMQ_Topic]  
[ActiveMQ专题5 ——JMS规范和落地产品][ActiveMQ_5 _JMS]  
[ActiveMQ专题6 ——ActiveMQ的Broker][ActiveMQ_6 _ActiveMQ_Broker]  
[ActiveMQ专题7 —— Spring整合ActiveMQ][ActiveMQ_7 _ Spring_ActiveMQ]  
[ActiveMQ专题8 —— SpringBoot整合ActiveMQ][ActiveMQ_8 _ SpringBoot_ActiveMQ]  
[ActiveMQ专题9 —— ActiveMQ的传输协议][ActiveMQ_9 _ ActiveMQ]  
[ActiveMQ专题10 —— ActiveMQ的存储和持久化][ActiveMQ_10 _ ActiveMQ]  
[ActiveMQ专题11 —— ActiveMQ的存储和持久化2 （续篇）][ActiveMQ_11 _ ActiveMQ_2]  
[ActiveMQ专题12—— ActiveMQ之zookeeper集群][ActiveMQ_12_ ActiveMQ_zookeeper]  
[ActiveMQ专题13—— ActiveMQ高级特性][ActiveMQ_13_ ActiveMQ]  
[ActiveMQ专题14—— ActiveMQ高级特性2][ActiveMQ_14_ ActiveMQ_2]  
$\\color\{red\} 完结！后续推上有关消息队列的面试题目 $

## 小总结 ##

知 道 的 越 多 ， 不 知 道 的 越 多 ， 希 望 对 你 有 帮 助 ！ \\color\{red\}知道的越多，不知道的越多，希望对你有帮助！ 知道的越多，不知道的越多，希望对你有帮助！

[2021030617505744.png]: /images/20221023/7e27edaf73564d3f95d1578cc78c344b.png
[Java_ActiveMQ_Topic]: https://blog.csdn.net/xiaozhegaa/article/details/114383509
[20210306175415171.png]: /images/20221023/05f588bd78e243399f8d91f1aeadbab7.png
[ActiveMQ_0 _]: https://blog.csdn.net/xiaozhegaa/article/details/114364355
[ActiveMQ_1 _]: https://blog.csdn.net/xiaozhegaa/article/details/114373971
[ActiveMQ_2 _ ActiveMQ_Linux]: https://blog.csdn.net/xiaozhegaa/article/details/114375436
[ActiveMQ_3 _Java_ActiveMQ_Queue]: https://blog.csdn.net/xiaozhegaa/article/details/114382174
[ActiveMQ_5 _JMS]: https://blog.csdn.net/xiaozhegaa/article/details/114388827
[ActiveMQ_6 _ActiveMQ_Broker]: https://blog.csdn.net/xiaozhegaa/article/details/114449488
[ActiveMQ_7 _ Spring_ActiveMQ]: https://blog.csdn.net/xiaozhegaa/article/details/114450231
[ActiveMQ_8 _ SpringBoot_ActiveMQ]: https://blog.csdn.net/xiaozhegaa/article/details/114451603
[ActiveMQ_9 _ ActiveMQ]: https://blog.csdn.net/xiaozhegaa/article/details/114452474
[ActiveMQ_10 _ ActiveMQ]: https://blog.csdn.net/xiaozhegaa/article/details/114457556
[ActiveMQ_11 _ ActiveMQ_2]: https://blog.csdn.net/xiaozhegaa/article/details/114461996
[ActiveMQ_12_ ActiveMQ_zookeeper]: https://blog.csdn.net/xiaozhegaa/article/details/114464565
[ActiveMQ_13_ ActiveMQ]: https://blog.csdn.net/xiaozhegaa/article/details/114494510
[ActiveMQ_14_ ActiveMQ_2]: https://blog.csdn.net/xiaozhegaa/article/details/114500737