MQ 系列之 ActiveMQ 基本使用

水深无声 2023-03-04 10:25 6阅读 0赞

![在这里插入图片描述][20200803170630818.png_pic_center]

## 1.1 ActiveMQ 的 API ##

### 1.1.1 ConnectionFactory ###

activemq-client 通过创建 ConnectionFactory 建立到 ActveMQ 的连接，连接工厂封装了一组连接配置参数，这组参数在配置 ActiveMQ 时已经定义，例如 brokerURL 参数，此参数传入的是 ActiveMQ 服务地址和端口，支持 openwire 协议的默认连接为 `tcp://localhost:61616`，支持 stomp 协议的默认连接为 `tcp://localhost:61613`。ActiveMQConnectionFactory 构造方法：  
  ♞ `ActiveMQConnectionFactory();`  
  ♞ `ActiveMQConnectionFactory(String brokerURL);`  
  ♞ `ActiveMQConnectionFactory(String userName, String password, String brokerURL);`  
  ♞ `ActiveMQConnectionFactory(String userName, String password, URI brokerURL);`  
  ♞ `ActiveMQConnectionFactory(URI brokerURL);`

### 1.1.2 Connection ###

在成功创建正确的 ConnectionFactory 后，下一步将是创建一个连接，它是 JMS 定义的一个接口。 ConnectionFactory 的静态方法 `createConnection()` 可以返回与底层消息传递系统进行通信的 Connection 实现。通常客户端只使用单一连接。根据 JMS 文档可知，Connection 的目的是“利用 JMS 提供者封装开放的连接”，以及表示“客户端与提供者服务例程之间的开放 TCP/IP 套接字”。该文档还指出 Connection 应该是进行客户端身份验证的地方，除了其他一些事项外，客户端还可以指定惟一标志符。当一个 Connection 被创建时，它的传输默认是关闭的，必须使用 start 方法开启。一个 Connection 可以建立一个或多个的 Session。当一个程序执行完成后，必须使用 close() 方法关闭之前创建的 Connection，否则 ActiveMQ 不能释放资源，关闭一个 Connection 后同样也关闭了 Session、MessageProducer 和 MessageConsumer。

### 1.1.3 Session ###

一旦从 ConnectionFactory 中获得一个 Connection，就必须从 Connection 中创建一个或者多个 Session。Session 是一个发送或接收消息的线程，可以使用 Session 创建 MessageProducer， MessageConsumer 和 Message。Session 可以被事务化，也可以不被事务化，通常，可以通过向 Connection 上的适当创建方法传递一个布尔参数对此进行设置。使用 `Session createSession(boolean transacted, int acknowledgeMode);` 创建 Session 对象，其中 transacted 为使用事务标识， acknowledgeMode 为签收模式。`AUTO_ACKNOWLEDGE = 1`：自动确认(一般选这个)；`CLIENT_ACKNOWLEDGE = 2`：客户端手动确认； `DUPS_OK_ACKNOWLEDGE = 3`：自动批量确认；`SESSION_TRANSACTED = 0`：事务提交并确认

### 1.1.4 Destination ###

Destination 是一个客户端用来指定生产消息目标和消费消息来源的对象。在 PTP 模式中，Destination 被称作 Queue 即队列；在 Pub/Sub 模式，Destination 被称作 Topic 即主题。在程序中可以使用多个 Queue 和 Topic。ActiveMQSession 中的方法：  
  ♞ `Queue createQueue(String queueName);`  
  ♞ `TemporaryQueue createTemporaryQueue();`  
  ♞ `Topic createTopic(String topicName);`  
  ♞ `TemporaryTopic createTemporaryTopic();`

### 1.1.5 Message ###

JMS 程序的最终目的是生产和消费的消息能被其他程序使用，JMS 的 Message 是一个既简单又不乏灵活性的基本格式，允许创建不同平台上符合非 JMS 程序格式的消息。Message 由以下几部分组成：消息头，属性和消息体。ActiveMQSession 方法：  
  ♞ `BlobMessage createBlobMessage(File file)`  
  ♞ `BlobMessage createBlobMessage(InputStream in)`  
  ♞ `BlobMessage createBlobMessage(URL url)`  
  ♞ `BlobMessage createBlobMessage(URL url, boolean deletedByBroker)`  
  ♞ `BytesMessage createBytesMessage()`  
  ♞ `MapMessage createMapMessage()`  
  ♞ `Message createMessage()`  
  ♞ `ObjectMessage createObjectMessage()`  
  ♞ `ObjectMessage createObjectMessage(Serializable object)`  
  ♞ `TextMessage createTextMessage()`  
  ♞ `TextMessage createTextMessage(String text)`

### 1.1.6 MessageProducer ###

MessageProducer 是消息生产者，是用来发送消息的，用来设定消息的优先级，签收的模式，以及消息的存活时间的设定。ActiveMQSession 中的方法：  
  ♞ `MessageProducer createProducer(Destination destination);`  
ActiveMQMessageProducer 中的方法：  
  ♞ `void send(Destination destination, Message message);`  
  ♞ `void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive);`  
  ♞ `void send(Message message);`  
  ♞ `void send(Message message, int deliveryMode, int priority, long timeToLive);`  
其中消息的目的地 `Destination`，发送的消息 `Message`，消息的模式 `DeliveryMode`，消息的优先级 `priority` 0-9 默认是 4，消息存活时间 `timeToLive` 存活时间单位是 ms 。

### 1.1.7 MessageConsumer ###

MessageConsumer 是消息消费者，是用来接收消息的。ActiveMQSession方法：  
  ♞ `MessageConsumer createConsumer(Destination destination);`  
  ♞ `MessageConsumer createConsumer(Destination destination, String messageSelector);`  
  ♞ `MessageConsumer createConsumer(Destination destination, String messageSelector, boolean noLocal);`  
  ♞ `TopicSubscriber createDurableSubscriber(Topic topic, String name);`  
  ♞ `TopicSubscriber createDurableSubscriber(Topic topic, String name, String messageSelector, boolean noLocal);`  
  其中 messageSelector 为消息选择器；noLocal 标志默认为 false，当设置为 true 时限制消费者只能接收和自己相同的连接(Connection)所发布的消息，此标志只适用于主题，不适用于队列；name 标识订阅主题所对应的订阅名称，持久订阅时需要设置此参数。

**☞ 消息的同步和异步接收**  
  消息的同步接收是指客户端主动去接收消息，客户端可以采用 MessageConsumer 的 receive 方法去接收下一个消息。消息的异步接收是指当消息到达时，ActiveMQ 主动通知客户端。客户端可以通过注册一个实现 MessageListener 接口的对象到 MessageConsumer。MessageListener 只有一个必须实现的方法 onMessage，它只接收一个参数，即 Message。在为每个发送到 Destination 的消息实现 onMessage 时，将调用该方法。ActiveMQMessageConsumer方法：  
  ♞ `Message receive()`  
  ♞ `Message receive(long timeout)`  
  ♞ `Message receiveNoWait()`  
实现 MessageListener 接口，每当消息到达时，ActiveMQ 会调用 MessageListener 中的 onMessage 函数。

**☞ 消息选择器**  
  JMS 提供了一种机制，使用它，消息服务可根据消息选择器中的标准来执行消息过滤。生产者可在消息中放入应用程序特有的属性，而消费者可使用基于这些属性的选择标准来表明对消息是否感兴趣。这就简化了客户端的工作，并避免了向不需要这些消息的消费者传送消息的开销。然而，它也使得处理选择标准的消息服务增加了一些额外开销。  
  消息选择器是用于 MessageConsumer 的过滤器，可以用来过滤传入消息的属性和消息头部分(但不过滤消息体)，并确定是否将实际消费该消息。按照 JMS 文档的说法，消息选择器是一些字符串，它们基于某种语法，而这种语法是 SQL-92 的子集。可以将消息选择器作为 MessageConsumer 创建的一部分。例如：`public final String SELECTOR = "JMSType = 'TOPIC_PUBLISHER'";`该选择器检查了传入消息的 JMSType 属性，并确定了这个属性的值是否等于 TOPIC\_PUBLISHER。如果相等，则消息被消费；如果不相等，那么消息会被忽略。

## 1.2 点对点模式 ##

### 1.2.1 概述 ###

点对点的模式主要建立在一个队列上面，当连接一个列队的时候，发送端不需要知道接收端是否正在接收，可以直接向 ActiveMQ 发送消息，发送的消息，将会先进入队列中，如果有接收端在监听，则会发向接收端，如果没有接收端接收，则会保存在 ActiveMQ 服务器，直到接收端接收消息，点对点的消息模式可以有多个发送端，多个接收端，但是一条消息，只会被一个接收端给接收到，哪个接收端先连上 ActiveMQ，则会先接收到，而后来的接收端则接收不到那条消息。  
![在这里插入图片描述][20200803163543839.png_pic_center]

### 1.2.2 消息生产者 ###

**☞ 添加依赖**

<dependency>
    	<groupId>org.apache.activemq</groupId>
    	<artifactId>activemq-client</artifactId>
    	<version>5.14.5</version>
    </dependency>
    
    
    <dependency>
    	<groupId>org.springframework</groupId>
    	<artifactId>spring-jms</artifactId>
    	<version>4.3.12.RELEASE</version>
    </dependency>

**☞ 生产消息**

/** * Created with IntelliJ IDEA. * * @author Demo_Null * @date 2020/8/3 * @description ActiveMQ 生产者 */
    public class ActiveMQProducer { 
        public static void main(String[] args) throws JMSException { 
            // 创建会话工厂对象 ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
    
            // 创建连接对象 Connection
            Connection connection = connectionFactory.createConnection();
    
            // 开启连接对象
            connection.start();
    
            // 获得会话 Session
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            // 创建消息
            TextMessage textMessage = session.createTextMessage();
            textMessage.setText("收到消息了吗？");
    
            // 指定消息发送地址和类型
            Destination destination = session.createQueue("test");
    
            // 创建消息发送对象
            MessageProducer producer = session.createProducer(destination);
    
            // 实现消息发送
            producer.send(textMessage);
    
            // 资源关闭
            session.close();
            connection.close();
        }
    }

![在这里插入图片描述][20200803150509248.png]  
![在这里插入图片描述][20200803150523481.png]

### 1.2.3 消息消费者 ###

**☞ 同步接收消息**

/** * Created with IntelliJ IDEA. * * @author Demo_Null * @date 2020/8/3 * @description ActiveMQ 消费者 */
    public class ActiveMQProducer { 
        public static void main(String[] args) throws JMSException { 
            // 创建 ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
    
            // 创建连接对象 Connection
            Connection connection = connectionFactory.createConnection();
    
            // 开启连接
            connection.start();
    
            // 创建会话对象
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            // 创建需要接收的消息地址
            Destination destination = session.createQueue("test");
    
    
            // 接收消息
            MessageConsumer consumer = session.createConsumer(destination);
    
            while (true){ 
                // 等待 10 秒，在 10 秒内一直处于接收消息状态
                Message message = consumer.receive(10000);
    
                if(null != message){ 
                 	// 将消息转换成 TextMessage
                    TextMessage textMessage = (TextMessage) message;
    
                    System.out.println("收到的内容是：" + textMessage.getText());
                } else { 
    				break;
    			}
            }
    
            // 关闭资源
            session.close();
            connection.close();
        }
    }

![在这里插入图片描述][20200803155938749.png]  
![在这里插入图片描述][20200803155952635.png]

**☞ 异步接收消息**

/** * Created with IntelliJ IDEA. * * @author Demo_Null * @date 2020/8/3 * @description ActiveMQ 消费者另外一种监听方式接收消息，该方法需要实现一个接口 MessageListener */
    public class ActiveMQConsumer { 
        public static void main(String[] args) throws Exception { 
            // 创建 ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
    
            // 创建连接对象 Connection
            Connection connection = connectionFactory.createConnection();
    
            // 开启连接
            connection.start();
    
            // 创建会话对象
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            // 创建需要接收的消息地址
            Destination destination = session.createQueue("test");
    
    
            // 接收消息
            MessageConsumer consumer = session.createConsumer(destination);
    
            //------------------监听创建开始-------------------
            // 监听消息，监听的方式是创建了一个线程
            consumer.setMessageListener(new MessageListener() { 
                @Override
                public void onMessage(Message message) { 
                    if(message instanceof TextMessage){ 
                        TextMessage textMessage = (TextMessage) message;
                        // 获取消息
                        try { 
                            System.out.println("消息内容：" + textMessage.getText());
                        } catch (JMSException e) { 
                            e.printStackTrace();
                        }
                    }
                }
            });
    
            // 为了防止主线程结束，这里录入键盘事件后再继续执行
            System.in.read();
            //------------------监听创建结束-------------------
    
            // 关闭资源
            session.close();
            connection.close();
        }
    }

## 1.3 发布/订阅模式 ##

### 1.3.1 概述 ###

发布/订阅消息模式是消息发送者发送消息到主题(topic)，而多个消息接收者监听这个主题；其中，消息发送者和接收者分别叫做发布者(publisher)和订阅者(subscriber)，对于发布者来说，它和所有的订阅者就构成了一个一对多的关系。消息生产者将消息(发布)到 topic 中，可以同时有多个消息消费者(订阅)消费该消息。和点对点方式不同，发布到 topic 的消息会被所有订阅者消费；当生产者发布消息时，不管是否有消费者，都不会保存消息；一定要先有消息的消费者，后有消息的生产者。  
![在这里插入图片描述][20200803163507305.png_pic_center]

### 1.3.2 订阅者 ###

**☞ 同步接收消息**

/** * Created with IntelliJ IDEA. * * @author Demo_Null * @date 2020/8/3 * @description 订阅者 */
    public class ActiveMQSubscriber02 { 
        public static void main(String[] args) throws Exception { 
            // 创建 ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
    
            // 创建连接对象 Connection
            Connection connection = connectionFactory.createConnection();
    
            // 开启连接
            connection.start();
    
            // 创建会话对象
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            // 创建需要接收的消息地址
            Topic topic = session.createTopic("topic");
    
    
            // 接收消息
            MessageConsumer consumer = session.createConsumer(test_topic);
    
            while (true){ 
                // 等待 10 秒，在 10 秒内一直处于接收消息状态
                Message message = consumer.receive(10000);
    
                if(null != message){ 
    	            // 等待 10 秒，在 10 秒内一直处于接收消息状态
    	            Message message = consumer.receive(10000);
    	
    	            if(null != message){ 
    	             	// 将消息转换成 TextMessage
    	                TextMessage textMessage = (TextMessage) message;
    	
    	                System.out.println("收到的内容是：" + textMessage.getText());
    	            } else { 
    					break;
    				}
                }
            }
    
            // 关闭资源
            session.close();
            connection.close();
        }
    }

![在这里插入图片描述][20200803162409202.png]  
![在这里插入图片描述][2020080316242441.png]

**☞ 异步接收消息**

/** * Created with IntelliJ IDEA. * * @author Demo_Null * @date 2020/8/3 * @description 订阅者另一种订阅消息方式 */
    public class ActiveMQSubscriber01 { 
        public static void main(String[] args) throws Exception { 
            // 创建 ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
    
            // 创建连接对象 Connection
            Connection connection = connectionFactory.createConnection();
    
            // 开启连接
            connection.start();
    
            // 创建会话对象
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            // 创建需要接收的消息地址
            Topic test_topic = session.createTopic("topic");
    
    
            // 接收消息
            MessageConsumer consumer = session.createConsumer(test_topic);
    
            //-------------关注这里开始----------------
            // 监听的方式是创建了一个线程
            consumer.setMessageListener(new MessageListener() { 
                @Override
                public void onMessage(Message message) { 
                    if(message instanceof TextMessage){ 
                        TextMessage textMessage = (TextMessage) message;
                        // 获取消息
                        try { 
                            System.out.println("消息是：" + textMessage.getText());
                        } catch (JMSException e) { 
                            e.printStackTrace();
                        }
                    }
                }
            });
    
            // 为了防止主线程结束，这里录入键盘事件后再继续执行
            System.in.read();
            //-------------关注这里结束----------------
    
            // 关闭资源
            session.close();
            connection.close();
        }
    }

### 1.3.3 发布者 ###

/** * Created with IntelliJ IDEA. * * @author gaohu9712@163.com * @date 2020/8/3 * @description 发布者 */
    public class ActiveMQPublisher { 
        public static void main(String[] args) throws JMSException { 
            // 创建会话工厂对象 ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
    
            // 创建连接对象 Connection
            Connection connection = connectionFactory.createConnection();
    
            // 开启连接对象
            connection.start();
    
            // 获得会话 Session
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            // 创建消息
            TextMessage textMessage = session.createTextMessage();
            textMessage.setText("发布消息");
    
            // 指定消息发送地址和类型
            Topic topic = session.createTopic("topic");
    
            // 创建消息发送对象
            MessageProducer producer = session.createProducer(topic);
    
            // 实现消息发送
            producer.send(textMessage);
    
            // 资源关闭
            session.close();
            connection.close();
        }
    }

![在这里插入图片描述][20200803162601256.png]  
![在这里插入图片描述][20200803162624645.png]  
![在这里插入图片描述][20200803162644768.png]

[20200803170630818.png_pic_center]: /images/20230208/8f4d60e15d624a14b023f5389da6a4b7.png
[20200803163543839.png_pic_center]: /images/20230208/6ec7e644d26a49849cf68630c097a5f3.png
[20200803150509248.png]: /images/20230208/544a2b71d64945ffae0bbf803e15c609.png
[20200803150523481.png]: /images/20230208/5b723dbd0cfa4e39b7c947303e6c9418.png
[20200803155938749.png]: /images/20230208/51e13dc8a19245bcb86b56a3d322cb3b.png
[20200803155952635.png]: /images/20230208/b479eb5a177d4450a9ead0d57a375088.png
[20200803163507305.png_pic_center]: /images/20230208/ab70b7fea5c64441bdec8b6b7ca597a2.png
[20200803162409202.png]: /images/20230208/15b8bc7219fe452681210f68d5b9d3de.png
[2020080316242441.png]: /images/20230208/83c2622646c842e799be4c193abf6109.png
[20200803162601256.png]: /images/20230208/9b182a3773174fa09d0fca3bb4e354a2.png
[20200803162624645.png]: /images/20230208/026ef7db11d044a1a77cd44110fe9297.png
[20200803162644768.png]: /images/20230208/281a76ff04b94eb49f6b942fe9449f5e.png