devbox
知新_RL
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Java编码实现Activemq通讯_java activemq

作者：知新_RL | 2024-04-08 08:59:04

赞

踩

java activemq

目录

一、IDEA建Maven工程

二、pom.xml

三、JMS编码总体架构

四、粗说目的地Destination——队列和主题

五、点对点消息传递域中，目的地被称为队列（queue）

六、在发布订阅消息传递域中，目的地被称为主题（topic）

七、topic和queue对比

一、IDEA建Maven工程

二、pom.xml

  1. <dependencies>
  2.         <!--  activemq  所需要的jar 包-->
  3.         <dependency>
  4.             <groupId>org.apache.activemq</groupId>
  5.             <artifactId>activemq-all</artifactId>
  6.             <version>5.15.9</version>
  7.         </dependency>
  8.         <!--  activemq 和 spring 整合的基础包 -->
  9.         <dependency>
  10.             <groupId>org.apache.xbean</groupId>
  11.             <artifactId>xbean-spring</artifactId>
  12.             <version>3.16</version>
  13.         </dependency>
  14.         <!--基础通用配置-->
  15.         <dependency>
  16.             <groupId>org.slf4j</groupId>
  17.             <artifactId>slf4j-api</artifactId>
  18.             <version>1.7.30</version>
  19.         </dependency>
  20.         <dependency>
  21.             <groupId>ch.qos.logback</groupId>
  22.             <artifactId>logback-classic</artifactId>
  23.             <version>1.2.11</version>
  24.         </dependency>
  25.         <dependency>
  26.             <groupId>org.projectlombok</groupId>
  27.             <artifactId>lombok</artifactId>
  28.             <version>1.16.16</version>
  29.         </dependency>
  30.         <dependency>
  31.             <groupId>junit</groupId>
  32.             <artifactId>junit</artifactId>
  33.             <version>4.12</version>
  34.             <scope>test</scope>
  35.         </dependency>
  36.     </dependencies>

三、JMS（Java平台上的专业技术规范）编码总体架构

 （1）回顾JDBC

 （2）JMS（Java平台上的专业技术规范）开发基本步骤

 

四、粗说目的地Destination——队列和主题

Destination是目的地。下面拿jvm和mq，做个对比。目的地，我们可以理解为是数据存储的地方。

Destination分为两种：队列和主题。

（1） 队列（queue——1对1）

 

（2）主题（topic——1对N）

 五、点对点消息传递域中，目的地被称为队列（queue）

（1）消息生产者

  1.     public static final String ACTIVEMQ_URL = "tcp://虚拟机ip:61616";
  2.     public static final String QUEUE_NAME = "queue01";
  3.  
  4.     public static void main(String[] args) throws JMSException {
  5.         //1.创建链接工厂,按照给定的url地址，采用默认的用户和密码
  6.         ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
  7.         //2.通过链接工厂，获得Connection并启动
  8.         Connection connection = activeMQConnectionFactory.createConnection();
  9.         connection.start();
  10.         //3.创建会话session
  11.         //3.1 两个参数 ①事务 ②签收
  12.         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  13.         //4.创建目的地（queue or topic）
  14.         Queue queue = session.createQueue(QUEUE_NAME);
  15.         //5.创建生产者
  16.         MessageProducer producer = session.createProducer(queue);
  17.         //6.通过MessageProducer生成3条消息到MQ队列中
  18.         for(int i = 1 ; i <= 3; i ++ ) {
  19.             TextMessage textMessage = session.createTextMessage("msg--" + i);
  20.             producer.send(textMessage);
  21.         }
  22.         producer.close();
  23.         session.close();
  24.         connection.close();
  25.  
  26.         System.out.println("消息发布到MQ完成……");
  27.     }

（2）控制台说明

回到ActiveMQ前端控制台可以看到如下界面，有三条消息

 

① Number Of Pending Messages：等待消费的消息，这个是未出队列的数量，公式=总接收数-总出队列数。

② Number Of Consumers：消费者数量，消费者端的消费者数量。

③ Messages Enqueued：进队消息数，进队列的总消息量，包括出队列的。这个数只增不减。

④ Messages Dequeued：出队消息数，可以理解为是消费者消费掉的数量。

总结：

当有一个消息进入这个队列时，等待消费的消息是1，进入队列的消息是1。

当消息消费后，等待消费的消息是0，进入队列的消息是1，出队列的消息是1。

当再来一条消息时，等待消费的消息是1，进入队列的消息就是2。

（3）消息消费者

① 方式一

  1.     public static final String ACTIVEMQ_URL = "tcp://虚拟机ip:61616";
  2.     public static final String QUEUE_NAME = "queue01";
  3. public static void main(String[] args) throws JMSException {
  4.         //1.创建链接工厂,按照给定的url地址，采用默认的用户和密码
  5.         ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
  6.         //2.通过链接工厂，获得Connection并启动
  7.         Connection connection = activeMQConnectionFactory.createConnection();
  8.         connection.start();
  9.         //3.创建会话session
  10.         //3.1 两个参数 ①事务 ②签收
  11.         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  12.         //4.创建目的地（queue or topic）
  13.         Queue queue = session.createQueue(QUEUE_NAME);
  14.         //5.创建消费者
  15.         MessageConsumer consumer = session.createConsumer(queue);
  16.         while(true) {
  17.             //2000ms timeout时间
  18.             //ActiveMQ有五种消息类型，这里发送时是TextMessage，接收时强转一下
  19.             /*同步阻赛方式(receive())
  20.               订阅者或接收者调Messaoeconsumerreceive()方法来接收消息，receive 方法在能够接收到消息之前(或超时之前)将一直阻塞*/
  21.             TextMessage textMessage = (TextMessage) consumer.receive(2000);
  22.             if(null != textMessage) {
  23.                 System.out.println("接收到消息：" + textMessage.getText());
  24.             } else {
  25.                 break;
  26.             }
  27.         }
  28.         consumer.close();
  29.         session.close();
  30.         connection.close();
  31.     }

 注意：MessageConsumer的receive方法如果选择的是无参类型的，即不设置超时时间（Timeout），那么上图的Number Of Consumers数量为1，因为消息消费完后receive方法会一直等待，程序一直运行，Consumer也就一直在线

② 方式二

  1. public static void main(String[] args) throws JMSException, IOException {
  2.         //1.创建链接工厂,按照给定的url地址，采用默认的用户和密码
  3.         ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
  4.         //2.通过链接工厂，获得Connection并启动
  5.         Connection connection = activeMQConnectionFactory.createConnection();
  6.         connection.start();
  7.         //3.创建会话session
  8.         //3.1 两个参数 ①事务 ②签收
  9.         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  10.         //4.创建目的地（queue or topic）
  11.         Queue queue = session.createQueue(QUEUE_NAME);
  12.         //5.创建消费者
  13.         MessageConsumer consumer = session.createConsumer(queue);
  14.         
  15.  
  16.         //通过监听方式读取消息
  17.         /*
  18.         异步非阻塞方式(监听器onMessage())
  19.             订阅者或接收者通过MessageConsumer的setMessageListener(MessageListener     
  20.             listener)注册一个消息监听器当消息到达之后，系统自动调用监听器MessageListener的 
  21.             onMessage(Message message)方法。
  22.         */
  23.         consumer.setMessageListener(new MessageListener() {
  24.             @Override
  25.             public void onMessage(Message message) {
  26.                 if(null != message && message instanceof TextMessage) {
  27.                     TextMessage textMessage = (TextMessage) message;
  28.                     try {
  29.                         System.out.println("监听到消息："+textMessage.getText());
  30.                     } catch (JMSException e) {
  31.                         e.printStackTrace();
  32.                     }
  33.                 }
  34.             }
  35.         });
  36.         //连接AvtiveMQ需要等待时间
  37.         System.in.read();
  38.         consumer.close();
  39.         session.close();
  40.         connection.close();
  41.     }

 

 ③ 消费者情况分析

  1.  一个消费者
  2. 两个消费者
    1. 先生成消息，先启动1号，在启动2号  结果：1号消费完所有消息，2号等待
    2. 先启动1，2号，再生成消息 结果：1，2号均分了消息，颇有些负载均衡的意味

④ 小总结

一、连接的基本步骤

 二、两种消费方式

① 同步阻塞方式(receive)

订阅者或接收者抵用MessageConsumer的receive()方法来接收消息，receive方法在能接收到消息之前（或超时之前）将一直阻塞。

② 异步非阻塞方式（监听器onMessage()）

订阅者或接收者通过MessageConsumer的setMessageListener(MessageListener listener)注册一个消息监听器，当消息到达之后，系统会自动调用监听器MessageListener的onMessage(Message message)方法。

三、队列特点

 

六、在发布订阅消息传递域中，目的地被称为主题（topic）

（1）topic介绍

在发布订阅消息传递域中，目的地被称为主题（topic）

发布/订阅消息传递域的特点如下：

① 生产者将消息发布到topic中，每个消息可以有多个消费者，属于1：N的关系；

② 生产者和消费者之间有时间上的相关性。订阅某一个主题的消费者只能消费自它订阅之后发布的消息。

③ 生产者生产时，topic不保存消息它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息，所以，一般先启动消费者再启动生产者。

默认情况下如上所述，但是JMS规范允许客户创建持久订阅，这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。一句话，好比我们的微信公众号订阅。

（2）发布主题生产者

  1.     public static final String ACTIVEMQ_URL = "tcp://虚拟机ip:61616";
  2.     public static final String TOPIC_NAME = "topic01";
  3. public static void main(String[] args) throws JMSException {
  4.         //1.创建链接工厂,按照给定的url地址，采用默认的用户和密码
  5.         ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
  6.         //2.通过链接工厂，获得Connection并启动
  7.         Connection connection = activeMQConnectionFactory.createConnection();
  8.         connection.start();
  9.         //3.创建会话session
  10.         //3.1 两个参数 ①事务 ②签收
  11.         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  12.         //4.创建目的地（queue or topic）
  13.         Topic topic = session.createTopic(TOPIC_NAME);
  14.         //5.创建生产者
  15.         MessageProducer producer = session.createProducer(topic);
  16.         //6.通过MessageProducer生成3条消息到MQ队列中
  17.         for(int i = 1 ; i <= 3; i ++ ) {
  18.             TextMessage textMessage = session.createTextMessage("topicMessage--" + i);
  19.             producer.send(textMessage);
  20.         }
  21.         producer.close();
  22.         session.close();
  23.         connection.close();
  24.  
  25.         System.out.println("topicMessage消息发布到MQ完成……");
  26.     }

（3）订阅主题消费者

  1.     public static final String ACTIVEMQ_URL = "tcp://虚拟机ip:61616";
  2.     public static final String TOPIC_NAME = "topic01";
  3. public static void main(String[] args) throws JMSException, IOException {
  4.         //1.创建链接工厂,按照给定的url地址，采用默认的用户和密码
  5.         ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
  6.         //2.通过链接工厂，获得Connection并启动
  7.         Connection connection = activeMQConnectionFactory.createConnection();
  8.         connection.start();
  9.         //3.创建会话session
  10.         //3.1 两个参数 ①事务 ②签收
  11.         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  12.         //4.创建目的地（queue or topic）
  13.         Topic topic = session.createTopic(TOPIC_NAME);
  14.         //5.创建消费者
  15.         MessageConsumer consumer = session.createConsumer(topic);
  16.  
  17.         consumer.setMessageListener(new MessageListener() {
  18.             @Override
  19.             public void onMessage(Message message) {
  20.                 if(null != message && message instanceof TextMessage) {
  21.                     TextMessage textMessage = (TextMessage) message;
  22.                     try {
  23.                         System.out.println("监听到消息："+textMessage.getText());
  24.                     } catch (JMSException e) {
  25.                         e.printStackTrace();
  26.                     }
  27.                 }
  28.             }
  29.         });
  30.         //连接AvtiveMQ需要等待时间
  31.         System.in.read();
  32.         consumer.close();
  33.         session.close();
  34.         connection.close();
  35.     }

（4）先启动消费者再启动生产者

① 启动消费者

 ② 启动生产者

 ③ 消费消息

 

④ 控制台

​​​​​​​

 七、topic和queue对比

 

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/知新_RL/article/detail/384286
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号