ActiveMQ的安装和使用

Posted 2021-01-06 huclouy

一、安装环境

　　jdk1.8

　　centos7 64系统

　　ActiveMQ5.13.0版本

二、安装

　　解压mq文件即可。

　　

三、启动和停止

　　启动命令：apache-activemq-5.13.0/bin/activemq start

　　

　　查看启动状态：apache-activemq-5.13.0/bin/activemq status

　　

　　停止：apache-activemq-5.13.0/bin/activemq stop

四、访问后台管理页面

　　URL：http://192.168.1.124:8161/admin

　　默认账号密码为admin/admin，默认端口为8161

 

五、ActiveMQ的使用方法

　　点对点：（一个只能对一个）

　　

　　发布/订阅：（一个对多个）

　　

　　添加所需要的依赖：

　　

.....
<activemq.version>5.13.0</activemq.version>
<spring.version>4.2.4.RELEASE</spring.version>
.....
<dependency>
       <groupId>org.apache.activemq</groupId>
       <artifactId>activemq-all</artifactId>
       <version>${activemq.version}</version>
       </dependency>
<dependency>
       <groupId>org.springframework</groupId>
       <artifactId>spring-jms</artifactId>
       <version>${spring.version}</version>
</dependency>
<dependency>
       <groupId>org.springframework</groupId>
       <artifactId>spring-context-support</artifactId>
       <version>${spring.version}</version>
</dependency>
......

　　5.1 　Queue使用　　

　　　　生产者：生产消息，发送端。

　　　　第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。

　　　　第二步：使用ConnectionFactory对象创建一个Connection对象。

　　　　第三步：开启连接，调用Connection对象的start方法。

　　　　第四步：使用Connection对象创建一个Session对象。

　　　　第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。

　　　　第六步：使用Session对象创建一个Producer对象。

　　　　第七步：创建一个Message对象，创建一个TextMessage对象。

　　　　第八步：使用Producer对象发送消息。

　　　　第九步：关闭资源。　　　

生产者：

public class Producer {

    @Test
    public void testQueueProduce() throws JMSException {
        // 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
        //brokerURL服务器的ip及端口号
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.124:61616");
        // 第二步：使用ConnectionFactory对象创建一个Connection对象。
        Connection connection = connectionFactory.createConnection();
        // 第三步：开启连接，调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。
        //第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
        //第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。
        //参数：队列的名称。
        Queue queue = session.createQueue("queue-test");
        // 第六步：使用Session对象创建一个Producer对象。
        MessageProducer producer = session.createProducer(queue);
        // 第七步：创建一个Message对象，创建一个TextMessage对象
        TextMessage textMessage = session.createTextMessage("this is my first mq message");
        // 第八步：使用Producer对象发送消息。
        producer.send(textMessage);
        //第九步：关闭资源
        producer.close();
        session.close();
        connection.close();
    }
}

　　运行该方法后，查看MQ控制台。

　　

 

消费者：

public class Consumer {

    @Test
    public void testQueueConsumer() throws JMSException, InterruptedException {
        //1.建立连接
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.124:61616");
        //2.获取连接
        Connection connection = connectionFactory.createConnection();
        //3.开启连接
        connection.start();
        //4.根据连接对象创建session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5.根据session创建Destination(目的地，Queue,Topic)
        Queue queue = session.createQueue("queue-test");
        //6.创建消费者
        MessageConsumer consumer = session.createConsumer(queue);
        //7.接收消息
        System.out.println("start");
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                if(message instanceof TextMessage) {
                    TextMessage message1 = (TextMessage) message;
                    try{
                        String text = message1.getText();
                        System.out.println("获取的消息为："+text);
                    }catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        System.out.println("end");
        Thread.sleep(10000);
        //8.关闭资源
        consumer.close();
        session.close();
        connection.close();
    }
}

执行后查看MQ控制台：

　　5.2 Topic的使用

　　　生产者：

@Test
    public void testTopicProducer() throws JMSException {
        // 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
        //brokerURL服务器的ip及端口号
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.124:61616");
        // 第二步：使用ConnectionFactory对象创建一个Connection对象。
        Connection connection = connectionFactory.createConnection();
        // 第三步：开启连接，调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。
        //第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
        //第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。
        //参数：队列的名称。
        Topic topic = session.createTopic("topic-test");
        // 第六步：使用Session对象创建一个Producer对象。
        MessageProducer producer = session.createProducer(topic);
        // 第七步：创建一个Message对象，创建一个TextMessage对象
//        TextMessage textMessage = session.createTextMessage("this is my first topic message");
        TextMessage message = new ActiveMQTextMessage();
        message.setText("my first topic message");
        // 第八步：使用Producer对象发送消息。
        producer.send(message);
        //第九步：关闭资源
        producer.close();
        session.close();
        connection.close();
    }

　　消费者：

@Test
    public void testTopicProducer() throws JMSException {
        // 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
        //brokerURL服务器的ip及端口号
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.124:61616");
        // 第二步：使用ConnectionFactory对象创建一个Connection对象。
        Connection connection = connectionFactory.createConnection();
        // 第三步：开启连接，调用Connection对象的start方法。
        connection.start();
        // 第四步：使用Connection对象创建一个Session对象。
        //第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
        //第二个参数：当第一个参数为false时，才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。
        //参数：队列的名称。
        Topic topic = session.createTopic("topic-test");
        // 第六步：使用Session对象创建一个Producer对象。
        MessageProducer producer = session.createProducer(topic);
        // 第七步：创建一个Message对象，创建一个TextMessage对象
//        TextMessage textMessage = session.createTextMessage("this is my first topic message");
        TextMessage message = new ActiveMQTextMessage();
        message.setText("my first topic message");
        // 第八步：使用Producer对象发送消息。
        producer.send(message);
        //第九步：关闭资源
        producer.close();
        session.close();
        connection.close();
    }

MQ控制台