ActiveMQ使用方法

- 日理万妓 2023-07-17 11:16 5阅读 0赞

# ActiveMQ使用方法 #

## 1.添加依赖包 ##

<dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-all</artifactId>
        <version>5.15.3</version>
    </dependency>

## 2.Queue ##

### 2.1 Product ###

**步骤如下:**

*  第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
 *  第二步：使用ConnectionFactory对象创建一个Connection对象。
 *  第三步：开启连接，调用Connection对象的start方法。
 *  第四步：使用Connection对象创建一个Session对象。
 *  第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。
 *  第六步：使用Session对象创建一个Producer对象。
 *  第七步：创建一个Message对象，创建一个TextMessage对象。
 *  第八步：使用Producer对象发送消息。
 *  第九步：关闭资源。

**具体代码如下：**

public static void main(String[] args) throws JMSException { 
    // 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
            ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
    // 第二步：使用ConnectionFactory对象创建一个Connection对象。
            Connection connection = factory.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("zhao-queue");
    // 第六步：使用Session对象创建一个Producer对象。
            MessageProducer producer = session.createProducer(queue);
    // 第七步：创建一个Message对象，创建一个TextMessage对象。
            TextMessage message = session.createTextMessage("hello world....");
    // 第八步：使用Producer对象发送消息。
            producer.send(message);
    // 第九步：关闭资源。
            producer.close();
            session.close();
            connection.close();
    
        }

**查看队列信息**  
![在这里插入图片描述][20200320223050726.png]  
![在这里插入图片描述][2020032022305566.png]

### 2.2 Consumer ###

**消费者：接收消息。**  
**步骤如下**

*  第一步：创建一个ConnectionFactory对象。
 *  第二步：从ConnectionFactory对象中获得一个Connection对象。
 *  第三步：开启连接。调用Connection对象的start方法。
 *  第四步：使用Connection对象创建一个Session对象。
 *  第五步：使用Session对象创建一个Destination对象。和发送端保持一致queue，并且队列的名称一致。
 *  第六步：使用Session对象创建一个Consumer对象。
 *  第七步：接收消息。
 *  第八步：打印消息。
 *  第九步：关闭资源

**实现代码**

public static Logger logger = Logger.getLogger(MyConsumer.class);
        public static void main(String[] args) throws JMSException, IOException { 
    // 第一步：创建一个ConnectionFactory对象。
            ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
    // 第二步：从ConnectionFactory对象中获得一个Connection对象。
            Connection connection = factory.createConnection();
    // 第三步：开启连接。调用Connection对象的start方法。
            connection.start();
    // 第四步：使用Connection对象创建一个Session对象。
            Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
    // 第五步：使用Session对象创建一个Destination对象。和发送端保持一致queue，并且队列的名称一致。
            Queue queue = session.createQueue("zhao-queue") ;
    // 第六步：使用Session对象创建一个Consumer对象。
            MessageConsumer consumer = session.createConsumer(queue);
    
    // 第七步：接收消息。
            consumer.setMessageListener(new MessageListener() { 
                @Override
                public void onMessage(Message message) { 
    
                    try { 
                        TextMessage textMessage = (TextMessage) message;
                        String text = null;
                        //获取消息内容
                        text = textMessage.getText();
                        // 第八步：打印消息。
                        logger.debug("message = "+text);
                        System.out.println("message = "+ text);
                    } catch (JMSException e) { 
                        e.printStackTrace();
                    }
    
                }
            });
            System.in.read();
    // 第九步：关闭资源
            consumer.close();
            session.close();
            connection.close();
        }

**可同时运行两个Consumer**  
![在这里插入图片描述][20200320223613640.png]  
每生产一个消息，只能被一个消费者消费一次。

## 3.Topic ##

### 3.1 Producer ###

**步骤如下：**

*  第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
 *  第二步：使用ConnectionFactory对象创建一个Connection对象。
 *  第三步：开启连接，调用Connection对象的start方法。
 *  第四步：使用Connection对象创建一个Session对象。
 *  第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Topic对象。
 *  第六步：使用Session对象创建一个Producer对象。
 *  第七步：创建一个Message对象，创建一个TextMessage对象。
 *  第八步：使用Producer对象发送消息。
 *  第九步：关闭资源。

**代码实现**

// 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
            ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
    // 第二步：使用ConnectionFactory对象创建一个Connection对象。
            Connection connection = factory.createConnection();
    // 第三步：开启连接，调用Connection对象的start方法。
            connection.start();
    // 第四步：使用Connection对象创建一个Session对象。
            Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
    // 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Topic对象。
            Topic topic = session.createTopic("test-topic");
    // 第六步：使用Session对象创建一个Producer对象。
             MessageProducer producer = session.createProducer(topic);
    // 第七步：创建一个Message对象，创建一个TextMessage对象。
            TextMessage message = session.createTextMessage("topic hello world ...");
    // 第八步：使用Producer对象发送消息。
            producer.send(message);
    // 第九步：关闭资源。
            producer.close();
            session.close();
            connection.close();
        }

**查看topic**  
![在这里插入图片描述][20200320224203451.png]

### 3.2 Consumer ###

**消费者：接收消息**  
**步骤如下:**

*  第一步：创建一个ConnectionFactory对象。
 *  第二步：从ConnectionFactory对象中获得一个Connection对象。
 *  第三步：开启连接。调用Connection对象的start方法。
 *  第四步：使用Connection对象创建一个Session对象。
 *  第五步：使用Session对象创建一个Destination对象。和发送端保持一致topic，并且话题的名称一致。
 *  第六步：使用Session对象创建一个Consumer对象。
 *  第七步：接收消息。
 *  第八步：打印消息。
 *  第九步：关闭资源

**代码**

public static Logger logger = Logger.getLogger(TopicConsumer.class);
        public static void main(String[] args) throws JMSException, IOException { 
    // 第一步：创建一个ConnectionFactory对象。
            ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
    // 第二步：从ConnectionFactory对象中获得一个Connection对象。
            Connection connection = factory.createConnection();
    // 第三步：开启连接。调用Connection对象的start方法。
            connection .start();
    // 第四步：使用Connection对象创建一个Session对象。
            Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
    // 第五步：使用Session对象创建一个Destination对象。和发送端保持一致topic，并且话题的名称一致。
            Topic topic = session.createTopic("test-topic");
    // 第六步：使用Session对象创建一个Consumer对象。
            MessageConsumer consumer = session.createConsumer(topic);
    // 第七步：接收消息。
            consumer.setMessageListener(new MessageListener() { 
                @Override
                public void onMessage(Message message) { 
                    try { 
                        TextMessage textMessage = (TextMessage) message;
                        String text = null;
                        text = textMessage.getText();
                        // 第八步：打印消息。
                        logger.debug(" message "+ text);
                        System.out.println("message "+ text);
                    }catch (Exception e){ 
                        e.printStackTrace();
                    }
                }
            });
            System.out.println("消费者 3 ");
          System.in.read();
    // 第九步：关闭资源
            consumer.close();
            session.close();
            connection.close();
    
        }

**查看topic**  
![在这里插入图片描述][20200320224457532.png]

## 总结 queue与topic的比较 ##

**可打开多个消费者进行测试！！！！**  
**queue形式**

*  queue是点对点的形式，product生产完消息后，可再打开consumer进行消费
 *  也同时用多个消费者进行消费，但每个消息只能被其中一个consumer消费一次。
 *  是要消息队列中有消息，消费者便可进行消费（监听）
 *  product与consumer的queue名字一定要相同

**topic形式**

*  topic使用发布/订阅模式，只有订阅的consumer才能消费。所有需要先订阅，即先打开consumer进行监听，不能直接从topic中取已经存在的消息。
 *  当多个consumer对同一个topic进行订阅后，每个consumer都可收到一份由product生产的消息
 *  订阅的consumer，有新消息发布时，consumer便可收到消息（监听）
 *  product与consumer的topic名字一定要相同

# 下期详讲ActiveMQ简单应用 #

[20200320223050726.png]: /images/20230528/6575a2f7ffba4947ab4777ab3d2deaba.png
[2020032022305566.png]: /images/20230528/33ffa625f0c04ed0a8a39cc087427720.png
[20200320223613640.png]: /images/20230528/2f2e734f5d05492b8a2755f57dab90df.png
[20200320224203451.png]: /images/20230528/3f9af90bc6484cee8c578accc207bcc1.png
[20200320224457532.png]: /images/20230528/926ae802c0cc479289bf0b732e36c385.png