ActiveMQ的Broker

一、什么是ActiveMQ的Broker

　　相当于一个ActiveMQ服务器实例.说白了,Broker其实就是实现了用代码的形式启动ActiveMQ,将MQ嵌入到Java代码中.以便随时需要随时启动,在用的时候再去启动这样能节省了资源,也保证了可用性.这种方式,我们实际开发中很少采用,因为他缺少太多了东西,比如:日志,数据存储等等.

　　

二、使用指定的配置文件来启动Broker

　　Broker:一个Broker就是Linux安装的一个activemq软件,我们只要把activemq.xml配置文件多复制几份,就能在一台服务器上启动多个Broker.

　　具体步骤:

　　1、到ActiveMQ的安装目录下面,进入到conf配置文件目录,拷贝一份activemq.xml配置文件(/usr/local/activemq/apache-activemq-5.15.5/conf)

// 当前位置,ActiveMQ安装位置的conf配置文件目录
[root@localhost conf]# pwd
/usr/local/activemq/apache-activemq-5.15.5/conf
// 拷贝activemq.xml并命名为activemq02.xml
[root@localhost conf]# cp activemq.xml activemq02.xml
[root@localhost conf]# ll
总用量 96
-rw-r--r--. 1 root root 5911 9月  23 19:01 activemq02.xml
-rw-r--r--. 1 root root 5911 8月   6 2018 activemq.xml
-rw-r--r--. 1 root root 1370 8月   6 2018 broker.ks
-rw-r--r--. 1 root root  592 8月   6 2018 broker-localhost.cert
-rw-r--r--. 1 root root  665 8月   6 2018 broker.ts
-rw-r--r--. 1 root root 1357 8月   6 2018 client.ks
-rw-r--r--. 1 root root  665 8月   6 2018 client.ts
-rw-r--r--. 1 root root 1172 8月   6 2018 credentials-enc.properties
-rw-r--r--. 1 root root 1121 8月   6 2018 credentials.properties
-rw-r--r--. 1 root root  962 8月   6 2018 groups.properties
-rw-r--r--. 1 root root 1011 8月   6 2018 java.security
-rw-r--r--. 1 root root 1087 8月   6 2018 jetty-realm.properties
-rw-r--r--. 1 root root 7795 8月   6 2018 jetty.xml
-rw-r--r--. 1 root root  965 8月   6 2018 jmx.access
-rw-r--r--. 1 root root  964 8月   6 2018 jmx.password
-rw-r--r--. 1 root root 3071 8月   6 2018 log4j.properties
-rw-r--r--. 1 root root 1207 8月   6 2018 logging.properties
-rw-r--r--. 1 root root 1016 8月   6 2018 login.config
-rw-r--r--. 1 root root  961 8月   6 2018 users.properties

　　

　　2、使用命令 ./activemq start xbean:file:/usr/local/activemq/apache-activemq-5.15.5/conf/activemq02.xml启动Broker

　　(我们之前使用命令 ./activemq start:实际上就是默认启动 /usr/local/activemq/apache-activemq-5.15.5/conf/activmq.xml配置文件,实际上完整的命令是:./activemq start xbean:file:/usr/local/activemq/apache-activemq-5.15.5/conf/activemq.xml)

　　

　　3、启动完成之后,生产者生产消息

　　

　　4、消费者消费消息

三、嵌入式Broker启动

　　1、EmbedBroker API

public class EmbedBroker {
    public static void main(String[] args) throws Exception {
        //ActiveMQ也支持在vm中通信基于嵌入的broker
        BrokerService brokerService = new BrokerService();
        brokerService.setPopulateJMSXUserID(true);
        brokerService.addConnector("tcp://localhost:61616");
        brokerService.start();
    }
}

　　启动控制台如下,可以看出本地确实启动了一个ActiveMQ实例

　　

　　2、生产者

public class JmsQueueProducer {
    public static final String BROKER_URL = "tcp://127.0.0.1:61616";
    public static final String QUEUE_NAME = "queue01";
    public static final String TEXT_MESSAGE_NAME = "textMessage";

    public static void main(String[] args) throws JMSException {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(QUEUE_NAME);
        MessageProducer producer = session.createProducer(queue);
        for (int i = 1; i < 9; i++) {
            TextMessage textMessage = session.createTextMessage(TEXT_MESSAGE_NAME + i);
            producer.send(queue, textMessage);
        }
        System.out.println("生产者成功发送消息至MQ QUEUE......");
        // 释放资源
        producer.close();
        session.close();
        connection.close();
    }
}

　　

　　3、消费者

public class JmsQueueConsumer {
    public static final String BROKER_URL = "tcp://127.0.0.1:61616";
    public static final String QUEUE_NAME = "queue01";

    public static void main(String[] args) throws IOException, JMSException {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(QUEUE_NAME);
        MessageConsumer consumer = session.createConsumer(queue);

        while (true) {
            Message message = consumer.receive(4 * 1000);
            if (message != null && message instanceof TextMessage) {
                TextMessage textMessage = (TextMessage) message;
                System.out.println("接收到的消息是:" + textMessage.getText());
            } else {
                break;
            }
        }
        consumer.close();
        session.close();
        connection.close();
    }
}