【MQ】ActiveMQ（二）——消息处理机制

心已赠人 2022-06-05 03:38 217阅读 0赞

# 一、前言 #

上文中，小编提到安装ActiveMQ，但是对于ActiveMQ中消息是用什么样的形式存储的？下面小编就向大家介绍一下。

# 二、消息类型 #

对于消息的传递有两种类型：

*  点对点的，即一个生产者和一个消费者一一对应；

![这里写图片描述][SouthEast]

*  发布/订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收。

![这里写图片描述][SouthEast 1]

JMS定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收以一些不同形式的数据，提供现有消息格式的一些级别的兼容性。

*  StreamMessage – Java原始值的数据流
 *  MapMessage–一套名称-值对
 *  TextMessage–一个字符串对象
 *  ObjectMessage–一个序列化的 Java对象
 *  BytesMessage–一个字节的数据流

# 三、点对点Queue #

## 3.1 Producer ##

生产者：生产消息，发送端。  
      把jar包添加到工程中。使用5.11.2版本的jar包。

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

@Test
        public void testQueueProducer() throws Exception {
            // 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
            //brokerURL服务器的ip及端口号
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168: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("test-queue");
            // 第六步：使用Session对象创建一个Producer对象。
            MessageProducer producer = session.createProducer(queue);
            // 第七步：创建一个Message对象，创建一个TextMessage对象。
            /*TextMessage message = new ActiveMQTextMessage(); message.setText("hello activeMq,this is my first test.");*/
            TextMessage textMessage = session.createTextMessage("hello activeMq,this is my first test.");
            // 第八步：使用Producer对象发送消息。
            producer.send(textMessage);
            // 第九步：关闭资源。
            producer.close();
            session.close();
            connection.close();
        }

## 3.2 12.1.2. Consumer ##

消费者：接收消息。

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

@Test
        public void testQueueConsumer() throws Exception {
            // 第一步：创建一个ConnectionFactory对象。
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
            // 第二步：从ConnectionFactory对象中获得一个Connection对象。
            Connection connection = connectionFactory.createConnection();
            // 第三步：开启连接。调用Connection对象的start方法。
            connection.start();
            // 第四步：使用Connection对象创建一个Session对象。
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 第五步：使用Session对象创建一个Destination对象。和发送端保持一致queue，并且队列的名称一致。
            Queue queue = session.createQueue("test-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();
                        // 第八步：打印消息。
                        System.out.println(text);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            //等待键盘输入
            System.in.read();
            // 第九步：关闭资源
            consumer.close();
            session.close();
            connection.close();
        }

## 四、发布订阅Topic ##

## 4.1 Producer ##

使用步骤：

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

@Test
        public void testTopicProducer() throws Exception {
            // 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
            // brokerURL服务器的ip及端口号
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168: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），此处创建一个topic对象。
            // 参数：话题的名称。
            Topic topic = session.createTopic("test-topic");
            // 第六步：使用Session对象创建一个Producer对象。
            MessageProducer producer = session.createProducer(topic);
            // 第七步：创建一个Message对象，创建一个TextMessage对象。
            /* * TextMessage message = new ActiveMQTextMessage(); message.setText( * "hello activeMq,this is my first test."); */
            TextMessage textMessage = session.createTextMessage("hello activeMq,this is my topic test");
            // 第八步：使用Producer对象发送消息。
            producer.send(textMessage);
            // 第九步：关闭资源。
            producer.close();
            session.close();
            connection.close();
        }

## 4.2 Consumer ##

消费者：接收消息。

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

@Test
        public void testTopicConsumer() throws Exception {
            // 第一步：创建一个ConnectionFactory对象。
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
            // 第二步：从ConnectionFactory对象中获得一个Connection对象。
            Connection connection = connectionFactory.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();
                        // 第八步：打印消息。
                        System.out.println(text);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            System.out.println("topic的消费端03。。。。。");
            // 等待键盘输入
            System.in.read();
            // 第九步：关闭资源
            consumer.close();
            session.close();
            connection.close();
        }

# 五、小结 #

通过分析和用不与spring结合的测试方法来更加深刻的了解mq在具体的使用过程中的步骤。

都是引入相关jar包后，创建连接工厂，工厂创建相关的连接对象，连接对象打开，创建相关的会话，会话发起目的地并且创建生产者或者消费者，然后对消息进行处理，最后关闭连接。

下一篇博客小编会向大家介绍如何使用spring管理ActiveMQ。

[SouthEast]: /images/20220605/a805d4a51c7b4d3ea898217808fb1772.png
[SouthEast 1]: /images/20220605/7a8176206d084932a536bb4658497fe5.png