-
ActiveMQ（一）
目录
一：消息队列 ( MQ ) 概述
1.消息队列的使用场景
展示业务场景：
我们在一个电商的系统中，当一个用户进行下单之后会进行跳转到多个步骤，这多个步骤都分配到一个微服务当中。
出现问题需求：
多个微服务之间进行交互依靠的是网络通信，我们知道网络始终是不可靠的，速度是相对比较慢的，也许还会出现网络动荡。
为了给予用户极好的体验，我们的需求就是：一下单就返回下单成功
解决需求：
我们使用的是MQ消息队列
步骤：
(1)用户下单之后，会发送一条消息给到消息队列MQ，这个消息包含着用户关于这个商品的信息，然后MQ会立即返回给用户一个下单成功的信息
然后用户完成支付的功能。这样极大的减少了用户的等待时间，给用户极好的用户体验 ！
(2)此时消息队列MQ中包含着1刚才用户下单的商品信息。后面多个下单的复杂过程对应的每一个微服务都会进行订阅这个消息队列。
(3)订阅这个消息队列之后，每一个订阅的微服务都会获取到消息队列中接收到的消息，然后每一个微服务再根据这个消息完成对下单过程的执行逻辑
总结：
使用消息队列之后，极大的完成了解耦。无需等到创建订单，扣减库存等等这些步骤都执行完毕，我们就可以给用户提示下单成功。
2.有Broker的MQ
这个流派通常有一台服务器作为Broker，所有的信息都通过它进行中转。生产者把消息发送给它就结束自己的任务了，Broker则把消息主动推送给消费者(或者消
费者主动轮询)
2.1 重Topic
2.2 轻Topic
3.无Broker的MQ
4.常用消息队列中间件的比较
二：ActiveMq的安装与测试服务
1.搞出一个activemq的虚拟机
2.
3.把在官网下载完成的activemq压缩包通过xftp上传到activemq目录
4.
5.
6.activemq依赖于Jdk环境，所以运行之前要进行注意看是否对应虚拟机具有JDK环境
使用xftp上传jdk
删除压缩包
7.检验activemq是否启动成功
8.查看activemq控制台
9.停止activemq服务
三：JMS框架
1.JMS介绍
2.JMS开发的基本步骤
(1) 首先消息生产者ConnectionFactory这个连接工厂会进行制造Connection连接对象
(2)创建完成连接对象之后，会进行和MQ的Broker进行开启一个Session会话
(3)Session会话开启成功之后，消息生产者就可以把生产的Message 传递到(send to)到我们的Broker中，由Destination队列进行接收
(4)假设有多个消息生产者的话，那么在Broker上会有多个Destination队列进行接收Message
(5)消息消费者也会通过ConnectionFactory这个连接工厂会进行制造Connection连接对象
(6)创建完成连接对象之后，会进行和MQ的Broker进行开启一个Session会话
(7)Session会话开启成功之后,消息消费者会进行receive from指定的Destination消息队列中的Message
(8)不同消息生产者或消费者只能从指定的Destination 队列中进行发送或接收Message
queue与topic的区别：
queue：
如果是队列的话，多个消费者等待接收这个队列中的Message，但是该Message最终只可以被一个消费者拿到 ！
topic：
topic则不同，只要订阅了这个主题topic的话，那么多个消费者都是可以拿到Message的 ！
四：JMS消息服务的实现
1.实现queue消息代码演示
1.导入依赖
org.apache.activemq activemq-all 5.17.0 org.apache.xbean xbean-spring 3.16 2.完成生产者的编写
代码：
public class ProducerDemo1 { //指明activemq的地址 public static final String URL = "tcp://192.168.204.134:61616" ; //指明destination目的地 public static final String QUEUE_NAME = "my_queue_1" ; public static void main(String[] args) throws Exception{ //1.获得连接工厂 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL) ; //2.获得连接对象 Connection connection = connectionFactory.createConnection(); //3.开启连接 connection.start(); //4.从连接对象中获得一个Session会话，该Session是MQ与消息生产者之间开启的会话 Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE) ; //5.会话对象进行指明Queue目的地，获取到队列对象 Queue queue = session.createQueue(QUEUE_NAME) ; //6.创建消息生产者 MessageProducer producer = session.createProducer(queue) ; //7.创建消息对象 TextMessage message = session.createTextMessage("hello activemq") ; //8.发送 producer.send(message) ; System.out.println("消息已发送"); //9.关闭连接 connection.close(); } }3.编写消息消费者
代码：
public class ConsumerDemo01 { //1.指明activemq的地址 public static final String URL = "tcp://192.168.204.134:61616" ; //2.指明destination目的地地址 public static final String QUEUE_NAME = "my_queue_1" ; public static void main(String[] args) throws JMSException { //3.获取连接工厂 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL) ; //4.获取连接 Connection connection = connectionFactory.createConnection(); //5.启动连接 connection.start(); //6.创建会话 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //7.会话Session对象进行指明目的地destination，获取得到队列对象 Queue queue = session.createQueue(QUEUE_NAME); //8.创建消费者对象 MessageConsumer consumer = session.createConsumer(queue) ; try { while (true) { //9.使用一个while循环，使得消费者一直在接收消息 Message message = consumer.receive() ; if (Objects.nonNull(message) && message instanceof TextMessage) { TextMessage textMessage = (TextMessage) message ; String text = textMessage.getText() ; System.out.println("接收到的消息内容为:" + text); } } } catch (JMSException e) { e.printStackTrace(); } finally { connection.close(); } } }2.消费者的同步阻塞方式
消费者的receive方式即为同步阻塞方式来接收消息。所谓的同步阻塞就是消费者获得消息后才能执行之后的业务逻辑，在接收到消息之前，一直是处于一个阻
塞等待的状态。
其中receive重载了两个方法：
1.一直阻塞等待
其中consumer.receive()方法在收到消息之前，一直处于阻塞等待的状态，之后的代码不会执行。直到收到消息后，才会执行之后的代码。
2.定义阻塞时间
3.同步阻塞与异步非阻塞的区别
同步阻塞：
(1) 小明给小红打电话其实就是一个同步阻塞的场景
(2) 同步：小明给小红打电话打成功的时候 二者必须都在线，这就是同步。
(3) 阻塞：小明给小红打电话的时候，在小红未同步的情况下，小明必须进行等待阻塞而不能立刻做完而去做其他的事情。
异步非阻塞：
(1)小明给小红发微信语音其实就是一个异步非阻塞的场景
(2) 异步：在网络正常的情况下，无论是否小红在线与否，小明都能给小红发微信语音发成功。
(3) 非阻塞：小明给小红发微信语音，由于是异步的，所以小明可以直接发送过去然后就去做自己其他的事情了。
4.消费者的异步非阻塞方式
消费者代码编写：
@SuppressWarnings({"all"}) public class ConsumerDemo02 { //1.指明activemq的地址 public static final String URL = "tcp://192.168.204.134:61616" ; //2.指明destination目的地地址 public static final String QUEUE_NAME = "my_queue_1" ; public static void main(String[] args) throws Exception { //3.获取连接工厂 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL) ; //4.获取连接 Connection connection = connectionFactory.createConnection(); //5.启动连接 connection.start(); //6.创建会话 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //7.会话Session对象进行指明目的地destination，获取得到队列对象 Queue queue = session.createQueue(QUEUE_NAME); //8.创建消费者对象 MessageConsumer consumer = session.createConsumer(queue) ; //9.接收消息的时候，通过设置监听器实现异步非阻塞 consumer.setMessageListener(message -> { //当接收到消息时，要执行的内容 if (Objects.nonNull(message) && message instanceof TextMessage) { TextMessage textMessage = (TextMessage) message ; //10.解析消息 String text = null; try { text = textMessage.getText(); } catch (JMSException e) { e.printStackTrace(); } System.out.println("接收到的消息为:" + text); } }); } }分析细节：
1.我们使用的是一个监听器进行监听，最终实现异步非阻塞的逻辑。
2.
9*
5.队列模式的特点
情况1：
(1) 先开启消息生产者，生产出三个Message。
(2) 开启消费者1进入消费，此时消费者1会消费完三个Message
(3) 当我们再次开启消费者2进入消费时，一个Message也消费不到，因为处于队列中的Message只可以被消费一次。
情况2：
(1) 先进行开启两个消息消费者进行等待
(2) 开启消息生产者进行创建Message
(3) 当有Message进入消息队列之后，当消息消费者订阅这个队列之后，会进行均摊读取，一人一条Message按照顺序读取。
测试结果一致：
记录一个queue与topic的细节：
queue模式：
当消息生产者生产Message然后传入到队列之后，即使没有一个消息消费者在线消费也没有关系。这个生产出的Message也不会消失，等到下次消息消费者接入之
后，也会接着进行消费之前生产传入的Message。
topic模式：
当消息生产者生产Message然后传入到队列之后，此时没有一个消息消费者在线消费。这个生产出的Message就作废了，等到消息消费者上线之后也不可以进行消
费这个情况下生产出的Message。
五、Topic模式
1.Topic介绍
一条Message会推送给多个消息消费者，这种一对多的关系：
2.代码实现
(1) topic的生产者实现
@SuppressWarnings({"all"}) public class producerdemo { //指明activemq的地址 public static final String URL = "tcp://192.168.204.134:61616" ; //指明destination目的地 public static final String TOPIC_NAME = "my_topic_1" ; public static void main(String[] args) throws Exception{ //1.获得连接工厂 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL) ; //2.获得连接对象 Connection connection = connectionFactory.createConnection(); //3.开启连接 connection.start(); //4.从连接对象中获得一个Session会话，该Session是MQ与消息生产者之间开启的会话 Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE) ; //5.会话对象进行指明destination目的地，获取到Topic对象 Topic topic = session.createTopic(TOPIC_NAME) ; //6.创建消息生产者 MessageProducer producer = session.createProducer(topic) ; for (int i = 0 ; i<10 ; i++) { //7.创建消息对象 TextMessage message = session.createTextMessage("hello topic" + i) ; //8.发送 producer.send(message) ; System.out.println("消息已发送"); } //9.关闭连接 connection.close(); } }(2)topic的消费者实现
@SuppressWarnings({"all"}) public class consumerdemo { //1.指明activemq的地址 public static final String URL = "tcp://192.168.204.134:61616" ; //2.指明destination目的地地址 public static final String TOPIC_NAME = "my_topic_1" ; public static void main(String[] args) throws JMSException { //1.创建ActiveMq的连接工厂 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL) ; //2.通过连接工厂获取Connection连接 Connection connection = connectionFactory.createConnection(); //3.启动连接 connection.start(); //4.创建Session会话 Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE) ; //5.创建Topic Topic topic = session.createTopic(TOPIC_NAME) ; //6.消费者通过Session会话连接Topic，然后准备开始消费 MessageConsumer consumer = session.createConsumer(topic) ; try { while (true) { //使用while循环并且使用receive方法表示时刻准备着消费 Message message = consumer.receive() ; if (Objects.nonNull(message) && message instanceof TextMessage) { String text = ((TextMessage) message).getText(); System.out.println("收到消息内容为: "+text); } } } catch (JMSException e) { e.printStackTrace(); } finally { connection.close(); } } }3.总结
topic模式这种模式是和之前的queue队列模式是类似的，都是一种连接消费者 然后让其消费消息的模式。
六：jMSMessage的组成部分
1.JMSMessage介绍
2.消息头
对于生产者生产出的消息的持久化模式分类：
(1) 如果其持久化模式是非持久化的，那么只会保留在内存中
(2) 其持久化模式是持久化的，那么会保留到磁盘文件中
代码演示部分：
3.消息体
MapMessage：
1.生产者
2.
4.消息属性
消息属性用于描述消息的额外信息。这个额外信息可以理解为用于区分消息的信息。比如给消息打标签，给消息分类等等，都可以使用消息属性来设置。
举个例子：
模拟一个需求场景：我们只进行接收author为Thor的消息生产者生产的消息，并且取出键值为name的值：
步骤演示：
(1) 消息生产者对应的设置：
(2)消息消费者的设置：
七：消息可靠性的实现
1.什么是消息可靠性
2.Queue队列的持久化
持久化是指的是消息是否被服务器进行存储。如果消息进行了持久化，则即使服务器宕机重启，消息在服务器重启后依然存在且可被消费。反之，则消息丢失。
持久化就是把消息数据设置到磁盘中，非持久化就是把消息数据设置到内存中
(1) 生产者设置非持久化：
(2) 持久化
3.Topic的发布订阅模式
(1)生产者发布Message到MQ的Broker上，接收的模式是topic模式
(2)多个消费者进行订阅这个topic，多个消费者都可以接收到生产者发布的Message
(3) 做到了生产者和消费者之间真正的解耦操作
消费者代码变化，生产者代码不变化：
4.Topic的持久化
(1) 对于topic模式来说，如果消息消费者没有进行订阅topic的broker，那么生产者发布的Message就近似一种失效的状态了。即使消费者重新进行订阅topic之后，也不可以接收到之前的Message。
(2)一旦消费者进行订阅之后，即使中间消费者下线没有直接接收到生产者发送到topic的消息也没关系，当消费者上线之后，就会一样的接收到Message的。
(3)假设说在消费者下线之后，生产者发布一Message，此时服务器宕机重启。
当消费者上线之后：
如果生产者发布的Message是持久化的，那么消费者上线之后就可以接收到这个Message
是非持久化的，。。。。。。。。。。不可以接收到这个Message
5.事务
生产者的事务
当开启事务之后，只有等到手动commit之后才可以说明把多个Message化为一个事务 然后才会进行真正的提交。否则Message一个都不会进行提交到topic
代码演示：
回滚操作：
消费者的事务：
代码实例演示：
细节记录：
6.ack签收
代码：
7.queue和topic总结
-
相关阅读:
大语言模型(LLM)综述(七)：大语言模型设计应用与未来方向
7-10 有关堆栈操作 Python
Android自定义控件属性函数
Android使用Banner框架实现轮播图
Mysql 主从复制与读写分离
计算机毕业设计Java宠物寄养预约系统(源码+系统+mysql数据库+lw文档)
【计算机网络】P2 性能指标
新华三辅导笔记 2023/10/9-2023/10/13
【PID优化】基于正余弦算法 (SCA)优化PID实现微型机器人系统位置控制附simulink模型和matlab代码
第一季：16JVM垃圾回收机制【Java面试题】
- 原文地址：https://blog.csdn.net/m0_61784000/article/details/127130932