ActiveMQ之队列和主题

小灰灰 2022-05-25 07:30 214阅读 0赞

ActiveMQ的使用分为两种，一种是嵌入式，即把ActiveMQ作为内嵌的JMS provider集成到Tomcat或其他Web服务器中；另一种是独立部署，即ActiveMQ单独部署，独立运行，通过ActiveMQ提供的API进行进程外访问。为了更有利于业务的解耦和工程部署结构的分离；也有利于对ActiveMQ进行配置定制和优化，可以采用独立部署的方式使用ActiveMQ。

JMS 支持两种截然不同的消息传送模型：PTP（即点对点模型）和Pub/Sub（即发布/订阅模型），分别称作：PTP Domain 和Pub/Sub Domain。  
PTP（使用Queue即队列目标）     消息从一个生产者传送至一个消费者。在此传送模型中，目标是一个队列。消息首先被传送至队列目标，然后根据队列传送策略，从该队列将消息传送至向此队列进行注册的某一个消费者，一次只传送一条消息。可以向队列目标发送消息的生产者的数量没有限制，但每条消息只能发送至、并由一个消费者成功使用。如果没有已经向队列目标注册的消费者，队列将保留它收到的消息，并在某个消费者向该队列进行注册时将消息传送给该消费者。  
Pub/Sub（使用Topic即主题目标）     消息从一个生产者传送至任意数量的消费者。在此传送模型中，目标是一个主题。消息首先被传送至主题目标，然后传送至所有已订阅此主题的活动消费者。可以向主题目标发送消息的生产者的数量没有限制，并且每个消息可以发送至任意数量的订阅消费者。主题目标也支持持久订阅的概念。持久订阅表示消费者已向主题目标进行注册，但在消息传送时此消费者可以处于非活动状态。当此消费者再次处于活动状态时，它将接收此信息。如果没有已经向主题目标注册的消费者，主题不保留其接收到的消息，除非有非活动消费者注册了持久订阅。

下面我们通过代码演示ActiveMQ对以上2种消息模式的使用方式。

引入activeMQ jar包

**\[html\]** [view plain][] [copy][view plain]

1.  <dependency>  
2.              <groupId>org.apache.activemq</groupId>  
3.              <artifactId>activemq-core</artifactId>  
4.              <version>5.8.0</version>  
5.          </dependency>

**一、队列模型**

在点对点的传输方式中，消息数据被持久化，每条消息都能被消费，没有监听QUEUE地址也能被消费，数据不会丢失，一对一的发布接受策略，保证数据完整。

生产者：

**\[java\]** [view plain][] [copy][view plain]

1.  package cn.slimsmart.activemq.demo.queue;  
2.    
3.  import javax.jms.Connection;  
4.  import javax.jms.ConnectionFactory;  
5.  import javax.jms.DeliveryMode;  
6.  import javax.jms.Destination;  
7.  import javax.jms.MessageProducer;  
8.  import javax.jms.Session;  
9.  import javax.jms.TextMessage;  
10.   
11. import org.apache.activemq.ActiveMQConnection;  
12. import org.apache.activemq.ActiveMQConnectionFactory;  
13.   
14. //队列  
15. public class Producer \{  
16.   
17.     public static void main(String\[\] args) throws Exception \{  
18.         // 连接工厂，JMS 用它创建连接  
19.         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.18.43:61616");  
20.         // 构造从工厂得到连接对象  
21.         Connection connection = connectionFactory.createConnection();  
22.         // 启动  
23.         connection.start();  
24.         // 获取操作连接 一个发送或接收消息的线程  
25.         Session session = connection.createSession(Boolean.TRUE, Session.AUTO\_ACKNOWLEDGE);  
26.         // 获取session注意参数值slimsmart.queue是一个queue  
27.         Destination destination = session.createQueue("slimsmart.queue.test");  
28.         // 得到消息生成者【发送者】  
29.         MessageProducer producer = session.createProducer(destination);  
30.         // 设置持久化，  
31.         producer.setDeliveryMode(DeliveryMode.PERSISTENT);  
32.         // 构造消息，并发送  
33.         sendMessage(session, producer);  
34.         //提交  
35.         session.commit();  
36.         session.close();  
37.         connection.close();  
38.     \}  
39.   
40.     public static void sendMessage(Session session, MessageProducer producer) throws Exception \{  
41.         for (int i = 1; i <= 5; i++) \{  
42.             TextMessage message = session.createTextMessage("发送的消息" + i);  
43.             // 发送消息到目的地方  
44.             System.out.println("发送消息：" + "发送的消息" + i);  
45.             producer.send(message);  
46.         \}  
47.     \}  
48.   
49. \}

消费者：

**\[java\]** [view plain][] [copy][view plain]

1.  package cn.slimsmart.activemq.demo.queue;  
2.    
3.  import javax.jms.Connection;  
4.  import javax.jms.ConnectionFactory;  
5.  import javax.jms.Destination;  
6.  import javax.jms.JMSException;  
7.  import javax.jms.Message;  
8.  import javax.jms.MessageConsumer;  
9.  import javax.jms.MessageListener;  
10. import javax.jms.Session;  
11. import javax.jms.TextMessage;  
12.   
13. import org.apache.activemq.ActiveMQConnection;  
14. import org.apache.activemq.ActiveMQConnectionFactory;  
15.   
16. public class Consumer \{  
17.   
18.     public static void main(String\[\] args) throws Exception \{  
19.         // 连接工厂，JMS 用它创建连接  
20.         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.18.43:61616");  
21.         // 构造从工厂得到连接对象  
22.         Connection connection = connectionFactory.createConnection();  
23.         // 启动  
24.         connection.start();  
25.         // 获取操作连接 一个发送或接收消息的线程  
26.         //第一个参数为true  标识开启事务，需要手工commit  
27.         final Session session = connection.createSession(Boolean.FALSE, Session.AUTO\_ACKNOWLEDGE);  
28.         // 获取session注意参数值slimsmart.queue是一个服务器的queue，星号\*匹配任意字符  
29.         Destination destination = session.createQueue("slimsmart.queue.test");  
30.         // 消费者，消息接收者  
31.         MessageConsumer consumer = session.createConsumer(destination);  
32.         consumer.setMessageListener( new MessageListener() \{  
33.             public void onMessage(Message message) \{  
34.                 TextMessage textMsg = (TextMessage)message;  
35.                 try \{  
36.                     System.out.println("收到消息" + textMsg.getText());  
37.                     //session.commit();  
38.                 \} catch (JMSException e) \{  
39.                     e.printStackTrace();  
40.                 \}  
41.             \}  
42.         \});  
43.         Thread.sleep(100000);  
44.         /\* 
45.         while (true) \{  
46.             //设置接收者接收消息的时间，为了便于测试，这里设定为100s 
47.             TextMessage message = (TextMessage) consumer.receive(100000); 
48.             if (null != message) \{  
49.                 System.out.println("收到消息" + message.getText()); 
50.             \} else \{  
51.                 break; 
52.             \} 
53.         \}\*/  
54.         session.close();  
55.         connection.close();  
56.     \}  
57. \}

**二、主题模型**  
在发布订阅消息方式中，消息是无状态的，不保证每条消息被消费，只有监听该TOPIC地址才能收到消息并消费，否则该消息将会丢失。一对多的发布接受策略，可以同时消费多个消息。

生产者：

**\[java\]** [view plain][] [copy][view plain]

1.  package cn.slimsmart.activemq.demo.topic;  
2.    
3.  import javax.jms.Connection;  
4.  import javax.jms.DeliveryMode;  
5.  import javax.jms.JMSException;  
6.  import javax.jms.MessageProducer;  
7.  import javax.jms.Session;  
8.  import javax.jms.TextMessage;  
9.  import javax.jms.Topic;  
10.   
11. import org.apache.activemq.ActiveMQConnectionFactory;  
12.   
13. public class Producer \{  
14.   
15.     public static void main(String\[\] args) throws JMSException \{  
16.         // 连接到ActiveMQ服务器  
17.         ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.18.43:61616");  
18.         Connection connection = factory.createConnection();  
19.         connection.start();  
20.         Session session = connection.createSession(Boolean.TRUE, Session.AUTO\_ACKNOWLEDGE);  
21.         // 创建主题  
22.         Topic topic = session.createTopic("slimsmart.topic.test");  
23.         MessageProducer producer = session.createProducer(topic);  
24.         // NON\_PERSISTENT 非持久化 PERSISTENT 持久化  
25.         producer.setDeliveryMode(DeliveryMode.PERSISTENT);  
26.         TextMessage message = session.createTextMessage();  
27.         message.setText("topic 消息。");  
28.         message.setStringProperty("property", "消息Property");  
29.         // 发布主题消息  
30.         producer.send(message);  
31.         System.out.println("Sent message: " + message.getText());  
32.         session.commit();  
33.         session.close();  
34.         connection.close();  
35.     \}  
36. \}

消费者：

package cn.slimsmart.activemq.demo.topic;  
      
    import javax.jms.Connection;  
    import javax.jms.JMSException;  
    import javax.jms.Message;  
    import javax.jms.MessageConsumer;  
    import javax.jms.MessageListener;  
    import javax.jms.Session;  
    import javax.jms.TextMessage;  
    import javax.jms.Topic;  
      
    import org.apache.activemq.ActiveMQConnectionFactory;  
      
    public class Consumer {  
      
        public static void main(String[] args) throws JMSException {  
            // 连接到ActiveMQ服务器  
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.18.43:61616");  
            Connection connection = factory.createConnection();  
            connection.start();  
            final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);  
            // 创建主题   
            Topic topic = session.createTopic("slimsmart.topic.test");  
            // 创建订阅  
            MessageConsumer consumer = session.createConsumer(topic);  
            consumer.setMessageListener(new MessageListener() {  
                // 订阅接收方法  
                public void onMessage(Message message) {  
                    TextMessage tm = (TextMessage) message;  
                    try {  
                        System.out.println("Received message: " + tm.getText()+":"+tm.getStringProperty("property"));  
                        session.commit();  
                    } catch (JMSException e) {  
                        e.printStackTrace();  
                    }  
                }  
            });  
        }  
    }

原文转自:[https://blog.csdn.net/zhu\_tianwei/article/details/46303023][https_blog.csdn.net_zhu_tianwei_article_details_46303023]

[view plain]: https://blog.csdn.net/zhu_tianwei/article/details/46303023#
[https_blog.csdn.net_zhu_tianwei_article_details_46303023]: https://blog.csdn.net/zhu_tianwei/article/details/46303023