目录
前言:为什么使用消息中间件?
异步通信
缓冲
解耦
前提:安装并启动activemq
一、点对点(point to point, queue)
1.1 创建maven项目
1.2 Pom依赖
1.2 JmsProduce 消息生产者
1.3 JmsConsumer1 消息消费者1
1.4 JmsConsumer2 消息消费者2
1.5 测试
1.5.1 启动两个消费者
1.5.2 启动生产者
二、发布/订阅(publish/subscribe,简称pub/sub,topic)
1.1 创建maven项目
1.2 Pom依赖
1.2 JmsProduce 消息生产者
1.3 JmsConsumer1 消息消费者1
1.4 JmsConsumer2 消息消费者2
1.5 测试
1.5.1 启动两个消费者
1.5.2 启动生产者发送6条消息
三、上述两种模式对比
前言:为什么使用消息中间件?
-
异步通信
异步化提升性能。有些业务不想也不需要立即处理消息。消息队列提供了异步处理机制,允许用户把一个消息放入队列,但并不立即处理它。想向队列中放入多少消息就放多少,然后在需要的时候再去处理它们。
-
缓冲
流量削峰。在任何重要的系统中,都会有需要不同的处理时间的元素。消息队列通过一个缓冲层来帮助任务最高效率的执行,该缓冲有助于控制和优化数据流经过系统的速度。以调节系统响应时间。
-
解耦
降低耦合度。降低工程间的强依赖程度,针对异构系统进行适配。在项目启动之初来预测将来项目会碰到什么需求,是极其困难的。通过消息系统在处理过程中间插入了一个隐含的、基于数据的接口层,两边的处理过程都要实现这一接口,当应用发生变化时,可以独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。
前提:安装并启动activemq
ActiveMQ介绍及linux下安装ActiveMQ-CSDN博客
一、点对点(point to point, queue)
消息生产者生产消息发送到queue中,然后消息消费者从queue中取出并且消费消息。
消息被消费以后,queue中不再有存储,所以消息消费者不可能消费到已经被消费的消息。Queue支持存在多个消费者,但是对一个消息而言,只会有一个消费者可以消费。
1.1 创建maven项目
1.2 Pom依赖
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.dolphin</groupId><artifactId>activemq_demo</artifactId><version>1.0-SNAPSHOT</version><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target></properties><dependencies><!-- activemq所需要的jar包配置 --><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-all</artifactId><version>5.15.9</version></dependency><dependency><groupId>org.apache.xbean</groupId><artifactId>xbean-spring</artifactId><version>3.16</version></dependency><!-- 下面是junit/log4j等基础通用配置 --><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.25</version></dependency><dependency><groupId>ch.qos.logback</groupId><artifactId>logback-classic</artifactId><version>1.2.3</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.16.18</version><scope>provided</scope></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version><scope>test</scope></dependency></dependencies>
</project>
1.3 JmsProduce 消息生产者
package com.dolphin;import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.jms.core.MessagePostProcessor;import javax.jms.*;public class JmsProduce {public static final String ACTIVEMQ_URL = "tcp://192.168.190.200:61616";public static final String QUEUE_NAME = "queue01";public static void main(String[] args) throws JMSException {//1 创建连接工厂,按照规定的url地址,采用默认用户名和密码 admin/adminActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2 通过连接工厂,获得链接connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3、创建会话session//两个参数,第一个叫事务/第二个叫签收Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4、创建目的地(具体是队列还是主题topic)Queue queue = session.createQueue(QUEUE_NAME);//5、创建消息的生产者MessageProducer messageProducer = session.createProducer(queue);//6、 通过使用messageProducer生产3条消息发送到MQ的队列里面for (int i = 1;i<=6;i++) {//7 创建消息TextMessage textMessage = session.createTextMessage("message---" + i);//理解为一个字符串//8 通过messageProducer发送给mqmessageProducer.send(textMessage);}//9 关闭资源messageProducer.close();session.close();connection.close();System.out.println("*****消息发布完成");}
}
1.4 JmsConsumer1 消息消费者1
package com.dolphin;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;
import java.io.IOException;public class JmsConsumer1 {public static final String ACTIVEMQ_URL = "tcp://192.168.190.200:61616";public static final String QUEUE_NAME = "queue01";public static void main(String[] args) throws JMSException, IOException {System.out.println("1号消费者");//1 创建连接工厂,按照规定的url地址,采用默认用户名和密码 admin/adminActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2 通过连接工厂,获得链接connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3、创建会话session//两个参数,第一个叫事务/第二个叫签收Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4、创建目的地(具体是队列还是主题topic)Queue queue = session.createQueue(QUEUE_NAME);//5、创建消息的消费者MessageConsumer messageConsumer = session.createConsumer(queue);//6、 通过使用messageProducer生产3条消息发送到MQ的队列里面/*** 两种消费方式(一)* 1 同步阻塞方式(receive())* 订阅者或接受者调用MessageConsumer的receive()方法来接收消息,receive方法在能够接收到消息之前(或者超时之前)将一直阻塞。while(true) {//超过4秒没有消息跳出循环关闭监听//TextMessage textMessage = (TextMessage) messageConsumer.receive(4000L);//没有消息也会一直阻塞等待消息TextMessage textMessage = (TextMessage) messageConsumer.receive();if (null != textMessage) {System.out.println("*****消费者接受到消息"+textMessage.getText());} else {break;}}//9 关闭资源messageConsumer.close();session.close();connection.close();System.out.println("*****消息消费完成");*//*** 两种消费方式(二)* 2 异步非阻塞方式(监听器onMessage())* 订阅者或接受者通过MessageConsumer的setMessageListener(MessageListener listener)注册一个消息监听器。* 当消息到达之后,系统自动调用监听器MessageListener的OnMessage(Message message)方法*/messageConsumer.setMessageListener(new MessageListener() {@Overridepublic void onMessage(Message message) {if (null != message && message instanceof TextMessage) {TextMessage textMessage = (TextMessage) message;try {System.out.println("*****消费者接受到消息"+textMessage.getText());} catch (JMSException e) {e.printStackTrace();}}}});System.in.read(); //进程处于运行状态messageConsumer.close();session.close();connection.close();System.out.println("*****消息消费完成");}
}
1.5 JmsConsumer2 消息消费者2
package com.dolphin;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;
import java.io.IOException;public class JmsConsumer1 {public static final String ACTIVEMQ_URL = "tcp://192.168.190.200:61616";public static final String QUEUE_NAME = "queue01";public static void main(String[] args) throws JMSException, IOException {System.out.println("1号消费者");//1 创建连接工厂,按照规定的url地址,采用默认用户名和密码 admin/adminActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2 通过连接工厂,获得链接connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3、创建会话session//两个参数,第一个叫事务/第二个叫签收Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4、创建目的地(具体是队列还是主题topic)Queue queue = session.createQueue(QUEUE_NAME);//5、创建消息的消费者MessageConsumer messageConsumer = session.createConsumer(queue);//6、 通过使用messageProducer生产3条消息发送到MQ的队列里面/*** 两种消费方式(一)* 1 同步阻塞方式(receive())* 订阅者或接受者调用MessageConsumer的receive()方法来接收消息,receive方法在能够接收到消息之前(或者超时之前)将一直阻塞。while(true) {//超过4秒没有消息跳出循环关闭监听//TextMessage textMessage = (TextMessage) messageConsumer.receive(4000L);//没有消息也会一直阻塞等待消息TextMessage textMessage = (TextMessage) messageConsumer.receive();if (null != textMessage) {System.out.println("*****消费者接受到消息"+textMessage.getText());} else {break;}}//9 关闭资源messageConsumer.close();session.close();connection.close();System.out.println("*****消息消费完成");*//*** 两种消费方式(二)* 2 异步非阻塞方式(监听器onMessage())* 订阅者或接受者通过MessageConsumer的setMessageListener(MessageListener listener)注册一个消息监听器。* 当消息到达之后,系统自动调用监听器MessageListener的OnMessage(Message message)方法*/messageConsumer.setMessageListener(new MessageListener() {@Overridepublic void onMessage(Message message) {if (null != message && message instanceof TextMessage) {TextMessage textMessage = (TextMessage) message;try {System.out.println("*****消费者接受到消息"+textMessage.getText());} catch (JMSException e) {e.printStackTrace();}}}});System.in.read(); //进程处于运行状态messageConsumer.close();session.close();connection.close();System.out.println("*****消息消费完成");}
}
1.6 测试
1.6.1 启动两个消费者
查看消费者数量2
1.6.2 启动生产者
这里生产了6个消息均被消费,且是轮询消费,结果如下
二、发布/订阅(publish/subscribe,简称pub/sub,topic)
发布订阅模式需要进行注册、订阅,根据注册消费对应的消息。多个生产者可以将消息写到同一个 Topic 中,多种消息可以被同一个消费者消费。一个生产者生产的消息,同样也可以被多个消费者消费,只要他们进行过消息订阅。
Queue支持存在多个消费者,对一个消息而言,可以有多个消费者共同消费。(如:微信公众号)
代码和上边案例一样,将queue改为topic,具体如下,这里直接在上边项目中修改
2.1 创建maven项目
2.2 Pom依赖
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.dolphin</groupId><artifactId>activemq_demo</artifactId><version>1.0-SNAPSHOT</version><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target></properties><dependencies><!-- activemq所需要的jar包配置 --><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-all</artifactId><version>5.15.9</version></dependency><dependency><groupId>org.apache.xbean</groupId><artifactId>xbean-spring</artifactId><version>3.16</version></dependency><!-- 下面是junit/log4j等基础通用配置 --><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.25</version></dependency><dependency><groupId>ch.qos.logback</groupId><artifactId>logback-classic</artifactId><version>1.2.3</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.16.18</version><scope>provided</scope></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version><scope>test</scope></dependency></dependencies>
</project>
2.3 JmsProduce 消息生产者
package com.dolphin;import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.jms.core.MessagePostProcessor;import javax.jms.*;public class JmsProduce_Topic {public static final String ACTIVEMQ_URL = "tcp://192.168.190.200:61616";public static final String TOPIC_NAME = "topic-weixin";public static void main(String[] args) throws JMSException {//1 创建连接工厂,按照规定的url地址,采用默认用户名和密码 admin/adminActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2 通过连接工厂,获得链接connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3、创建会话session//两个参数,第一个叫事务/第二个叫签收Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4、创建目的地(具体是队列还是主题topic)Topic topic = session.createTopic(TOPIC_NAME);//5、创建消息的生产者MessageProducer messageProducer = session.createProducer(topic);//6、 通过使用messageProducer生产3条消息发送到MQ的队列里面for (int i = 1;i<=6;i++) {//7 创建消息TextMessage textMessage = session.createTextMessage("TOPIC_NAME---" + i);//理解为一个字符串//8 通过messageProducer发送给mqmessageProducer.send(textMessage);}//9 关闭资源messageProducer.close();session.close();connection.close();System.out.println("*****消息发布完成");}
}
2.4 JmsConsumer1 消息消费者1
package com.dolphin;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;
import java.io.IOException;public class JmsConsumer1 {public static final String ACTIVEMQ_URL = "tcp://192.168.190.200:61616";public static final String TOPIC_NAME = "topic-weixin";public static void main(String[] args) throws JMSException, IOException {System.out.println("1号消费者");//1 创建连接工厂,按照规定的url地址,采用默认用户名和密码 admin/adminActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2 通过连接工厂,获得链接connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3、创建会话session//两个参数,第一个叫事务/第二个叫签收Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4、创建目的地(具体是队列还是主题topic)Topic topic = session.createTopic(TOPIC_NAME);//5、创建消息的消费者MessageConsumer messageConsumer = session.createConsumer(topic);//6、 通过使用messageProducer生产3条消息发送到MQ的队列里面messageConsumer.setMessageListener(message -> {if (null != message && message instanceof TextMessage) {TextMessage textMessage = (TextMessage) message;try {System.out.println("*****1号消费者接受到消息:"+textMessage.getText());} catch (JMSException e) {e.printStackTrace();}}});System.in.read(); //进程处于运行状态messageConsumer.close();session.close();connection.close();System.out.println("*****消息消费完成");}
}
2.5 JmsConsumer2 消息消费者2
package com.dolphin;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;
import java.io.IOException;public class JmsConsumer2 {public static final String ACTIVEMQ_URL = "tcp://192.168.190.200:61616";public static final String TOPIC_NAME = "topic-weixin";public static void main(String[] args) throws JMSException, IOException {System.out.println("2号消费者");//1 创建连接工厂,按照规定的url地址,采用默认用户名和密码 admin/adminActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2 通过连接工厂,获得链接connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3、创建会话session//两个参数,第一个叫事务/第二个叫签收Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4、创建目的地(具体是队列还是主题topic)Topic topic = session.createTopic(TOPIC_NAME);//5、创建消息的消费者MessageConsumer messageConsumer = session.createConsumer(topic);//6、 通过使用messageProducer生产3条消息发送到MQ的队列里面messageConsumer.setMessageListener(message -> {if (null != message && message instanceof TextMessage) {TextMessage textMessage = (TextMessage) message;try {System.out.println("*****2号消费者接受到消息:"+textMessage.getText());} catch (JMSException e) {e.printStackTrace();}}});System.in.read(); //进程处于运行状态messageConsumer.close();session.close();connection.close();System.out.println("*****消息消费完成");}
}
2.6 测试
2.6.1 启动两个消费者
查看消费者数量2
2.6.2 启动生产者发送6条消息
三、上述两种模式对比
比较项目 | Topic模式队列 | Queue模式队列 |
---|---|---|
工作模式 | “订阅-发布模式”,如果当前没有订阅者,消息将会被丢弃。如果有多个订阅者,那么这些订阅者都会收到消息 | “负载均衡”模式,如果当前没有消费者,消息也不会丢弃;如果有多个消费者,那么一条消息也只会发送给其中一个消费者,并且要求消费者ack信息 |
有无状态 | 无状态 | Queue数据默认会在mq服务器上以文件形式保存,比如ActiveMQ一般保存在$AMQ_HOME\data\kr-store\data下面。也可以配置成DB存储 |
传递完整性 | 如果没有订阅者,消息会被丢弃 | 消息不会丢弃 |
处理效率 | 由于消息要按照订阅者的数量进行复制,所以处理性能会随着订阅者的增加而明显降低,并且还要结合不同消息协议自身的性能差异 | 由于一条消息只发给一个消费者,所以就算消费者再多,性能也不会有明显降低。当然不同消息协议的具体性能也是有差异的 |
四、消息持久化设置
1. queue默认是持久化
设置非持久化
message..setJMSDeliveryMode(DeliveryMode.NON_PERSISTENT); 生产者配置
2. topic默认非持久化
设置为持久化
2.1 生产者创建连接之前设置持久化
2.2 生产者创建连接之前设置持久化,并对应修改代码
启动消费者端这里会显示在线的持久化订阅者