干货｜SpringBootJMS（ActiveMQ）API实践应用详解

[[350573]]

 前言

Active是一种开源的，实现了JMS1.1规范的，面向消息(MOM)的中间件，为应用程序提供高效的、可扩展的、稳定的和安全的企业级消息通信。AC-tiveMQ使用Apache提供的授权，任何人都可以对其实现代码进行修改。

ActiveMQ的设计目标是提供标准的，面向消息的，能够跨越多语言和多系统的应用集成消息通信中间件。ActiveMQ实现了JMS标准并提供了很多附加的特性。本文将带大家详细介绍ActiveMQ的API的使用。

1. JMS的概念?

「什么是JMS呢：」

  • JMS---------JAVA Message Service
  • JAVA的消息服务，是sun公司提供的接口，只是一个规范，这个规范就类似于JDBC是一样的，使用的时候是需要当前规范的实现产品的。

「JMS能干什么呢：」

  • 能够将信息发布到目的地
  • 可以从目的地来消费这个消息

2、两种通信模型

「队列的通信概念：」

  • 特点：当我们同一个队列有多个消费者的时候，多个消费者的数据之和才是原来队列中的所有数据
  • 队列的通信模型最大的适用场景：流量的消峰，高并发的处理

「主题的通信模型：」

  • 特点：当我们队列有多个消费者的时候，那么这多个消费者消费到的数据是一样的
  • 主题消费者通信模型的适用场景：微服务下服务之间的异步通信

3. MQ的实现产品

「实现产品：」

  • ActiveMQ
  • RabbitMQ
  • RockerMQ
  • Kafka(这个设计的初衷是做分布式的日志的，后来因为日志有严格的顺序问题，这个时候人们就用Kafka来做消息队列了)

4、JMS中常见的名词

「常见的名词：」

  • ActiveMQConnectionFactory：这个是创建连接的工厂
  • ConnectionFactory：连接的工厂
  • Connection：连接JAVA对MQ的一个连接
  • Destination：目的地
  • 生产者(Producer)
  • 消费者(Consumer)
  • Session：会话(每一次对MQ的操作都称为一次会话)
  • Queue：队列
  • Topic：主题

5、什么是消息队列

「消息队列简单的说就是用来存放临时数据的地方：」

  • 生产者----------->存储介质上
  • 消费者----------->存储介质上

「消息队列类似于快递公司：」

  • 你可以将东西交给快递公司
  • 目标人也可以从快递公司去取东西

6. ActiveMQ是什么

「含义：」

  • ActiveMQ就是一个JMS的实现产品，它能够实现JMS下的所有功能

7、ActiveMQ能干什么

「主要作用：」

  • 流量消峰处理
  • 微服务下模块的异步通信
  • 处理高并发下的订单
  • 处理第三方平台的高并发
  • 协助消息表可以完成分布式事务的最终一致性

8、ActiveMQ的安装

「ActiveMQ的安装和配置：」

 
 
 
    
  
  
  
  1.  1、官网下载Linux版的ActiveMQ（最新版本为5.13.4） 
    2.   
  
  
  2.        https://activemq.apache.org/download.html 
    3.   
  
  
  3.  
    4.   
  
  
  4.        2、解压安装 
    5.   
  
  
  5.       tar -zxvf apache-activemq-5.13.4-bin.tar.gz 
    6.   
  
  
  6.  
    7.   
  
  
  7.        3、配置（这里采用默认配置，无需修改） 
    8.   
  
  
  8.       vim /usr/lical/activemq-1/conf/activemq.xml 
    9.   
  
  
  9.  
    10.   
  
  
  10.        4、启动 
    11.   
  
  
  11.       cd /usr/local/activemq-1/bin 
    12.   
  
  
  12. ./activemq start 
    13.   
  
  
  13.  
    14.   
  
  
  14.       5、打开管理界面（管理界面可以查看并管理所有队列及消息） 
    15.   
  
  
  15.          http://192.168.1.100:8161 
    16.   
  
  
  16.  
    17.   
  
  
  17.        启动成功后，可以浏览 http://localhost:8161/admin/ 
    18.   
  
  
  18.        默认用户名、密码：admin/admin 
    19.   
  
  
  19.        管理界面是用jetty做容器的，如果想修改管理界面的端口，可以编辑../conf/jetty.xml，找到下面这一段： 
    20.   
  
  
  20.        
    21.   
  
  
  21.      
    22.   
  
  
  22.      
    23.   
  
  
  23.      
    24.   
  
  
  24.       
    25.   
  
  
  25.        用户名/密码是在 ../conf/jetty-realm.properties 里，比如要增加一个管理员jimmy/123456，可参考下面修改： 
    26.   
  
  
  26.        1 
    27.   
  
  
  27. 2 
    28.   
  
  
  28. 3admin: admin, admin 
    29.   
  
  
  29. jimmy: 123456, admin 
    30.   
  
  
  30. user: user, user 
    31.   
  
  
  31.        注：管理界面有一个小坑，ActiveMQ 5.13.2与jdk1.8兼容性有点问题，如果使用jdk1.8，管理界面进入Queues标签页时，偶尔会报错，但是并不影响消息正常收发，只是无法从界面上查看队列情况，如果出现该问题，可将jdk版本降至1.7，同时最好清空data目录下的所有数据，再重启activemq即可。 
    32.

9. ActiveMQ的API的使用

「AcatveMQ的API使用：」

 

  • 队列的使用(生产者)
 
 
 
    
  
  
  
  1. package com.qy.mq.queue; 
    2.   
  
  
  2.  
    3.   
  
  
  3. import org.apache.activemq.ActiveMQConnectionFactory; 
    4.   
  
  
  4. import org.apache.activemq.Message; 
    5.   
  
  
  5.  
    6.   
  
  
  6. import javax.jms.*; 
    7.   
  
  
  7.  
    8.   
  
  
  8. /** 
    9.   
  
  
  9.  * @Auther: qianyu 
    10.   
  
  
  10.  * @Date: 2020/11/04 14:12 
    11.   
  
  
  11.  * @Description:生产者 
    12.   
  
  
  12.  */ 
    13.   
  
  
  13. public class Producer { 
    14.   
  
  
  14.  
    15.   
  
  
  15.     //准备发布的这个地址 
    16.   
  
  
  16.     private  static final String PATH="tcp://10.7.182.87:61616"; 
    17.   
  
  
  17.     //ActiveMQ下的用户名 
    18.   
  
  
  18.     private static final String userName="admin"; 
    19.   
  
  
  19.     //ActiveMQ下的密码 
    20.   
  
  
  20.     private static final String password="admin"; 
    21.   
  
  
  21.  
    22.   
  
  
  22.     public static void main(String[] args) throws JMSException { 
    23.   
  
  
  23.         //第一步：创建连接的工厂 
    24.   
  
  
  24.         ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(userName, password, PATH); 
    25.   
  
  
  25.         //通过这个工厂获取连接 
    26.   
  
  
  26.         Connection connection = activeMQConnectionFactory.createConnection(); 
    27.   
  
  
  27.         //第三步：打开这个连接 
    28.   
  
  
  28.         connection.start(); 
    29.   
  
  
  29.         //第四步：创建操作MQ的这个会话 
    30.   
  
  
  30.         /** 
    31.   
  
  
  31.          * 第一个参数：是否使用事务 
    32.   
  
  
  32.          * 第二个参数：客户端的应答模式 
    33.   
  
  
  33.          *     第一种：自动应答 
    34.   
  
  
  34.          *     第二种：客户端手动应答 
    35.   
  
  
  35.          */ 
    36.   
  
  
  36.         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 
    37.   
  
  
  37.         //需要发送消息的目的地(queue操作的是队列) 
    38.   
  
  
  38.         Destination destination=session.createQueue("wqqq"); 
    39.   
  
  
  39.         //生产者来生产这个消息 
    40.   
  
  
  40.         //要有生产者 
    41.   
  
  
  41.         MessageProducer messageProducer = session.createProducer(destination); 
    42.   
  
  
  42.  
    43.   
  
  
  43.         //发送很多的消息到消息队列中去 
    44.   
  
  
  44. //        for (int i=0;i<100;i++){ 
    45.   
  
  
  45.             //需要准备发送的消息 
    46.   
  
  
  46. //            TextMessage textMessage = session.createTextMessage("我是浅羽:"+i); 
    47.   
  
  
  47.             //研究消息的类型 
    48.   
  
  
  48.  
    49.   
  
  
  49.           /*  BytesMessage bytesMessage = session.createBytesMessage(); 
    50.   
  
  
  50.             bytesMessage.setByteProperty("www",new Byte("123")); 
    51.   
  
  
  51.  
    52.   
  
  
  52.             //接下来就可以发送消息了 
    53.   
  
  
  53.             messageProducer.send(bytesMessage);*/ 
    54.   
  
  
  54.  
    55.   
  
  
  55.         //创建map类型的message 
    56.   
  
  
  56.         /*MapMessage mapMessage = session.createMapMessage(); 
    57.   
  
  
  57.         mapMessage.setInt("www1",123); 
    58.   
  
  
  58.  
    59.   
  
  
  59.         messageProducer.send(mapMessage);*/ 
    60.   
  
  
  60.  
    61.   
  
  
  61.         ObjectMessage objectMessage = session.createObjectMessage(new User(1, "qianyu", "123")); 
    62.   
  
  
  62.         messageProducer.send(objectMessage); 
    63.   
  
  
  63.  
    64.   
  
  
  64. //        } 
    65.   
  
  
  65.     } 
    66.   
  
  
  66.  
    67.   
  
  
  67. } 
    68.
  • 队列的使用（消费者）
 
 
 
    
  
  
  
  1. package com.qy.mq.queue; 
    2.   
  
  
  2.  
    3.   
  
  
  3. import org.apache.activemq.ActiveMQConnectionFactory; 
    4.   
  
  
  4.  
    5.   
  
  
  5. import javax.jms.*; 
    6.   
  
  
  6. import java.io.Serializable; 
    7.   
  
  
  7.  
    8.   
  
  
  8. /** 
    9.   
  
  
  9.  * @Auther: qianyu 
    10.   
  
  
  10.  * @Date: 2020/11/04 14:13 
    11.   
  
  
  11.  * @Description:消费者 
    12.   
  
  
  12.  */ 
    13.   
  
  
  13. public class Consumer { 
    14.   
  
  
  14.  
    15.   
  
  
  15.     //准备发布的这个地址 
    16.   
  
  
  16.     private  static final String PATH="tcp://10.7.182.87:61616"; 
    17.   
  
  
  17.     //ActiveMQ下的用户名 
    18.   
  
  
  18.     private static final String userName="admin"; 
    19.   
  
  
  19.     //ActiveMQ下的密码 
    20.   
  
  
  20.     private static final String password="admin"; 
    21.   
  
  
  21.  
    22.   
  
  
  22.     public static void main(String[] args) throws JMSException { 
    23.   
  
  
  23.         //第一步：创建连接的工厂 
    24.   
  
  
  24.         ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(userName, password, PATH); 
    25.   
  
  
  25.         //通过这个工厂获取连接 
    26.   
  
  
  26.         Connection connection = activeMQConnectionFactory.createConnection(); 
    27.   
  
  
  27.         //第三步：打开这个连接 
    28.   
  
  
  28.         connection.start(); 
    29.   
  
  
  29.         //第四步：创建操作MQ的这个会话 
    30.   
  
  
  30.         /** 
    31.   
  
  
  31.          * 第一个参数：是否使用事务 
    32.   
  
  
  32.          * 第二个参数：客户端的应答模式 
    33.   
  
  
  33.          *     第一种：自动应答 
    34.   
  
  
  34.          *     第二种：客户端手动应答 
    35.   
  
  
  35.          */ 
    36.   
  
  
  36.         Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); 
    37.   
  
  
  37.         //需要发送消息的目的地(queue操作的是队列) 
    38.   
  
  
  38.         Destination destination=session.createQueue("wqqq"); 
    39.   
  
  
  39.         //创建我们的消费者了 
    40.   
  
  
  40.         MessageConsumer messageConsumer = session.createConsumer(destination); 
    41.   
  
  
  41.         //接下来就可以接收我们的消息了 
    42.   
  
  
  42.         //Message message = messageConsumer.receive(); 
    43.   
  
  
  43.  
    44.   
  
  
  44.         //接收消息并指定这个超时的时间 
    45.   
  
  
  45. //      Message message = messageConsumer.receive(5000); 
    46.   
  
  
  46.  
    47.   
  
  
  47.         //接收消息没有就不等待了 直接over了  不接收了 
    48.   
  
  
  48. //      Message message = messageConsumer.receiveNoWait(); 
    49.   
  
  
  49.  
    50.   
  
  
  50.         //给定当前的路径设置监听器 
    51.   
  
  
  51.         messageConsumer.setMessageListener(new MessageListener() { 
    52.   
  
  
  52.             @Override 
    53.   
  
  
  53.             public void onMessage(Message message) { 
    54.   
  
  
  54.  
    55.   
  
  
  55.                /* BytesMessage bytesMessage= (BytesMessage) message; 
    56.   
  
  
  56.  
    57.   
  
  
  57.                 try { 
    58.   
  
  
  58.                     System.out.println("获取到的数据是:"+bytesMessage.getByteProperty("www")); 
    59.   
  
  
  59.                 } catch (JMSException e) { 
    60.   
  
  
  60.                     e.printStackTrace(); 
    61.   
  
  
  61.                 }*/ 
    62.   
  
  
  62.  
    63.   
  
  
  63.  
    64.   
  
  
  64.                /*MapMessage mapMessage= (MapMessage) message; 
    65.   
  
  
  65.  
    66.   
  
  
  66.                 try { 
    67.   
  
  
  67.                     System.out.println("获取到的数据是:"+mapMessage.getInt("www1")); 
    68.   
  
  
  68.                 } catch (JMSException e) { 
    69.   
  
  
  69.                     e.printStackTrace(); 
    70.   
  
  
  70.                 }*/ 
    71.   
  
  
  71.  
    72.   
  
  
  72.                //测试对象类型的消息的发送和接收 
    73.   
  
  
  73.                ObjectMessage objectMessage= (ObjectMessage) message; 
    74.   
  
  
  74.                 try { 
    75.   
  
  
  75.                     User user = (User) objectMessage.getObject(); 
    76.   
  
  
  76.  
    77.   
  
  
  77.                     System.out.println("接收到的数据是:"+user); 
    78.   
  
  
  78.                 } catch (JMSException e) { 
    79.   
  
  
  79.                     e.printStackTrace(); 
    80.   
  
  
  80.                 } 
    81.   
  
  
  81.  
    82.   
  
  
  82.  
    83.   
  
  
  83.               /*  //我们知道是一个字符串类型的消息 
    84.   
  
  
  84.                 TextMessage textMessage= (TextMessage) message; 
    85.   
  
  
  85.                 //接下来就可以打印这个消息了 
    86.   
  
  
  86.                 try { 
    87.   
  
  
  87.                     System.out.println("消费者1---接收到的消息是："+textMessage.getText()); 
    88.   
  
  
  88.                 } catch (JMSException e) { 
    89.   
  
  
  89.                     e.printStackTrace(); 
    90.   
  
  
  90.                 }*/ 
    91.   
  
  
  91.  
    92.   
  
  
  92.                 try { 
    93.   
  
  
  93.                     //这句话就表示的是客户端来手动的进行应答 
    94.   
  
  
  94.                     message.acknowledge(); 
    95.   
  
  
  95.                 } catch (JMSException e) { 
    96.   
  
  
  96.                     e.printStackTrace(); 
    97.   
  
  
  97.                 } 
    98.   
  
  
  98.  
    99.   
  
  
  99.             } 
    100.   
  
  
  100.         }); 
    101.   
  
  
  101.     } 
    102.   
  
  
  102. } 
    103.
  • 主题模型的生产者
 
 
 
    
  
  
  
  1. package com.qy.mq.topic; 
    2.   
  
  
  2.  
    3.   
  
  
  3. import org.apache.activemq.ActiveMQConnectionFactory; 
    4.   
  
  
  4.  
    5.   
  
  
  5. import javax.jms.*; 
    6.   
  
  
  6.  
    7.   
  
  
  7. /** 
    8.   
  
  
  8.  * @Auther: qianyu 
    9.   
  
  
  9.  * @Date: 2020/11/04 15:17 
    10.   
  
  
  10.  * @Description: 
    11.   
  
  
  11.  */ 
    12.   
  
  
  12. public class Producer { 
    13.   
  
  
  13.     //准备发布的这个地址 
    14.   
  
  
  14.     private  static final String PATH="tcp://10.7.182.87:61616"; 
    15.   
  
  
  15.     //ActiveMQ下的用户名 
    16.   
  
  
  16.     private static final String userName="admin"; 
    17.   
  
  
  17.     //ActiveMQ下的密码 
    18.   
  
  
  18.     private static final String password="admin"; 
    19.   
  
  
  19.  
    20.   
  
  
  20.     public static void main(String[] args) throws JMSException { 
    21.   
  
  
  21.         //第一步：创建连接的工厂 
    22.   
  
  
  22.         ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(userName, password, PATH); 
    23.   
  
  
  23.         //通过这个工厂获取连接 
    24.   
  
  
  24.         Connection connection = activeMQConnectionFactory.createConnection(); 
    25.   
  
  
  25.         //第三步：打开这个连接 
    26.   
  
  
  26.         connection.start(); 
    27.   
  
  
  27.         //第四步：创建操作MQ的这个会话 
    28.   
  
  
  28.         /** 
    29.   
  
  
  29.          * 第一个参数：是否使用事务 
    30.   
  
  
  30.          * 第二个参数：客户端的应答模式 
    31.   
  
  
  31.          *     第一种：自动应答 
    32.   
  
  
  32.          *     第二种：客户端手动应答 
    33.   
  
  
  33.          */ 
    34.   
  
  
  34.         Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); 
    35.   
  
  
  35.         //需要发送消息的目的地(下面创建的就应该是主题模型的地址) 
    36.   
  
  
  36.         Destination destination=session.createTopic("topic222"); 
    37.   
  
  
  37.         //生产者来生产这个消息 
    38.   
  
  
  38.         //要有生产者 
    39.   
  
  
  39.         MessageProducer messageProducer = session.createProducer(destination); 
    40.   
  
  
  40.  
    41.   
  
  
  41.         //发送很多的消息到消息队列中去 
    42.   
  
  
  42.         for (int i=0;i<100;i++){ 
    43.   
  
  
  43.             //需要准备发送的消息 
    44.   
  
  
  44.             TextMessage textMessage = session.createTextMessage("我是浅羽:"+i); 
    45.   
  
  
  45.             //接下来就可以发送消息了 
    46.   
  
  
  46.             messageProducer.send(textMessage); 
    47.   
  
  
  47.         } 
    48.   
  
  
  48.     } 
    49.   
  
  
  49. } 
    50.
  • 主题模型的消费者
 
 
 
    
  
  
  
  1. package com.qy.mq.topic; 
    2.   
  
  
  2.  
    3.   
  
  
  3. import org.apache.activemq.ActiveMQConnectionFactory; 
    4.   
  
  
  4.  
    5.   
  
  
  5. import javax.jms.*; 
    6.   
  
  
  6.  
    7.   
  
  
  7. /** 
    8.   
  
  
  8.  * @Auther: qianyu 
    9.   
  
  
  9.  * @Date: 2020/11/04 15:19 
    10.   
  
  
  10.  * @Description: 
    11.   
  
  
  11.  */ 
    12.   
  
  
  12. public class Consumer { 
    13.   
  
  
  13.  
    14.   
  
  
  14.     //准备发布的这个地址 
    15.   
  
  
  15.     private  static final String PATH="tcp://10.7.182.87:61616"; 
    16.   
  
  
  16.     //ActiveMQ下的用户名 
    17.   
  
  
  17.     private static final String userName="admin"; 
    18.   
  
  
  18.     //ActiveMQ下的密码 
    19.   
  
  
  19.     private static final String password="admin"; 
    20.   
  
  
  20.  
    21.   
  
  
  21.     public static void main(String[] args) throws JMSException { 
    22.   
  
  
  22.         //第一步：创建连接的工厂 
    23.   
  
  
  23.         ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(userName, password, PATH); 
    24.   
  
  
  24.         //通过这个工厂获取连接 
    25.   
  
  
  25.         Connection connection = activeMQConnectionFactory.createConnection(); 
    26.   
  
  
  26.         //第三步：打开这个连接 
    27.   
  
  
  27.         connection.start(); 
    28.   
  
  
  28.         //第四步：创建操作MQ的这个会话 
    29.   
  
  
  29.         /** 
    30.   
  
  
  30.          * 第一个参数：是否使用事务 
    31.   
  
  
  31.          * 第二个参数：客户端的应答模式 
    32.   
  
  
  32.          *     第一种：自动应答 
    33.   
  
  
  33.          *     第二种：客户端手动应答 
    34.   
  
  
  34.          */ 
    35.   
  
  
  35.         Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); 
    36.   
  
  
  36.         //需要发送消息的目的地(queue操作的是队列) 
    37.   
  
  
  37.         Destination destination=session.createTopic("topic222"); 
    38.   
  
  
  38.         //创建我们的消费者了 
    39.   
  
  
  39.         MessageConsumer messageConsumer = session.createConsumer(destination); 
    40.   
  
  
  40.         //接下来就可以接收我们的消息了 
    41.   
  
  
  41.  
    42.   
  
  
  42.         //给定当前的路径设置监听器 
    43.   
  
  
  43.         messageConsumer.setMessageListener(new MessageListener() { 
    44.   
  
  
  44.             @Override 
    45.   
  
  
  45.             public void onMessage(Message message) { 
    46.   
  
  
  46.  
    47.   
  
  
  47.                 //我们知道是一个字符串类型的消息 
    48.   
  
  
  48.                 TextMessage textMessage= (TextMessage) message; 
    49.   
  
  
  49.                 //接下来就可以打印这个消息了 
    50.   
  
  
  50.                 try { 
    51.   
  
  
  51.                     System.out.println("消费者1---接收到的消息是："+textMessage.getText()); 
    52.   
  
  
  52.                 } catch (JMSException e) { 
    53.   
  
  
  53.                     e.printStackTrace(); 
    54.   
  
  
  54.                 } 
    55.   
  
  
  55.  
    56.   
  
  
  56.                 try { 
    57.   
  
  
  57.                     message.acknowledge(); 
    58.   
  
  
  58.                 } catch (JMSException e) { 
    59.   
  
  
  59.                     e.printStackTrace(); 
    60.   
  
  
  60.                 } 
    61.   
  
  
  61.  
    62.   
  
  
  62.             } 
    63.   
  
  
  63.         }); 
    64.   
  
  
  64.     } 
    65.   
  
  
  65. } 
    66.

本篇关于ActiveMQ的介绍就先到这里结束了，后续会出更多关于ActiveMQ系列更多文章，谢谢大家支持！

分享名称：干货｜SpringBootJMS（ActiveMQ）API实践应用详解
文章位置：http://www.shufengxianlan.com/qtweb/news2/382302.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

服务器托管知识

分类信息网站

云主机　　 　温江服务器托管　　 　成都广告设计公司　　 　抖音代运营　　 　成都康明斯发电机　　 　成都封阳台　　 　产品包装设计　　 　网站建设公司　　 　禾巨建站　　 　眉山服务器托管　　 　成都响应式网站建设公司　　 　软文推广　　 　免备案空间　　 　达州网站设计　　 　巴彦网站建设　　 　域名申请　　 　成都网络推广公司　　 　网站制作　　 　德阳机房　　 　外贸营销网站建设