马云的网站是谁建设的,广西住房和城乡建设官方网站,深圳seo优化外包,巢湖网站建设费用目录
前言#xff1a;为什么使用消息中间件#xff1f;
异步通信
缓冲
解耦
前提#xff1a;安装并启动activemq
一、点对点#xff08;point to point#xff0c; queue#xff09;
1.1 创建maven项目
1.2 Pom依赖
1.2 JmsProduce 消息生产者
1.3 JmsConsumer…目录
前言为什么使用消息中间件
异步通信
缓冲
解耦
前提安装并启动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/subtopic
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 version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcom.dolphin/groupIdartifactIdactivemq_demo/artifactIdversion1.0-SNAPSHOT/versionpropertiesproject.build.sourceEncodingUTF-8/project.build.sourceEncodingmaven.compiler.source1.8/maven.compiler.sourcemaven.compiler.target1.8/maven.compiler.target/propertiesdependencies!-- activemq所需要的jar包配置 --dependencygroupIdorg.apache.activemq/groupIdartifactIdactivemq-all/artifactIdversion5.15.9/version/dependencydependencygroupIdorg.apache.xbean/groupIdartifactIdxbean-spring/artifactIdversion3.16/version/dependency!-- 下面是junit/log4j等基础通用配置 --dependencygroupIdorg.slf4j/groupIdartifactIdslf4j-api/artifactIdversion1.7.25/version/dependencydependencygroupIdch.qos.logback/groupIdartifactIdlogback-classic/artifactIdversion1.2.3/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.16.18/versionscopeprovided/scope/dependencydependencygroupIdjunit/groupIdartifactIdjunit/artifactIdversion4.12/versionscopetest/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、创建目的地具体是队列还是主题topicQueue queue session.createQueue(QUEUE_NAME);//5、创建消息的生产者MessageProducer messageProducer session.createProducer(queue);//6、 通过使用messageProducer生产3条消息发送到MQ的队列里面for (int i 1;i6;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、创建目的地具体是队列还是主题topicQueue 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、创建目的地具体是队列还是主题topicQueue 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/subtopic
发布订阅模式需要进行注册、订阅根据注册消费对应的消息。多个生产者可以将消息写到同一个 Topic 中多种消息可以被同一个消费者消费。一个生产者生产的消息同样也可以被多个消费者消费只要他们进行过消息订阅。
Queue支持存在多个消费者对一个消息而言可以有多个消费者共同消费。如微信公众号 代码和上边案例一样将queue改为topic具体如下这里直接在上边项目中修改
2.1 创建maven项目 2.2 Pom依赖
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcom.dolphin/groupIdartifactIdactivemq_demo/artifactIdversion1.0-SNAPSHOT/versionpropertiesproject.build.sourceEncodingUTF-8/project.build.sourceEncodingmaven.compiler.source1.8/maven.compiler.sourcemaven.compiler.target1.8/maven.compiler.target/propertiesdependencies!-- activemq所需要的jar包配置 --dependencygroupIdorg.apache.activemq/groupIdartifactIdactivemq-all/artifactIdversion5.15.9/version/dependencydependencygroupIdorg.apache.xbean/groupIdartifactIdxbean-spring/artifactIdversion3.16/version/dependency!-- 下面是junit/log4j等基础通用配置 --dependencygroupIdorg.slf4j/groupIdartifactIdslf4j-api/artifactIdversion1.7.25/version/dependencydependencygroupIdch.qos.logback/groupIdartifactIdlogback-classic/artifactIdversion1.2.3/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.16.18/versionscopeprovided/scope/dependencydependencygroupIdjunit/groupIdartifactIdjunit/artifactIdversion4.12/versionscopetest/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、创建目的地具体是队列还是主题topicTopic topic session.createTopic(TOPIC_NAME);//5、创建消息的生产者MessageProducer messageProducer session.createProducer(topic);//6、 通过使用messageProducer生产3条消息发送到MQ的队列里面for (int i 1;i6;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、创建目的地具体是队列还是主题topicTopic 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、创建目的地具体是队列还是主题topicTopic 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 生产者创建连接之前设置持久化并对应修改代码 启动消费者端这里会显示在线的持久化订阅者
