ActivityMQ基础
一、消息中间件概述
什么是消息中间件？与传统的传输通讯有什么区别？答：异步，无需等待，消息存放在队列里面。调用别人的接口时候，返回的是同步还是异步？答：发送请求/响应是同步的。
什么是同步？什么是异步？
当A项目调用B接口时，B接口有延迟，会产生什么场景？答：1、A会一直等待，等待B影响给我，http有设置超时时间。2、同步接口中，如果网络延迟，可能会产生重复提交接口产生的重复提交如何解决？采用token(令牌）验证。防止机器模拟请求采用 验证码的方式验证、A调用B如果B没有及时响应。A项目默认有3次重试补偿机制。将该信息存放在日志表（补偿表）中,再使用定时job每天晚上健康检查数据，可以手动补偿B责任：处理幂等性问题。
请求与响应 = 同步请求方式同步请求方式：缺点：1、会阻塞、超时、还会发生重复提交导致数据不一致。
二、为什么要使用消息中间件
为什么要使用消息中间件？答：消息中间件可以解决高并发、两种通讯方式：点对点通讯、发布订阅、异步通讯(无需等待)
生产者：发送消息，提供接口的，主要向队列中发送消息队列：存放消息地址消息：发送的报文信息消费者：调用接口的，主要从队列中获取消息1、第一种通讯方式（消息模型） 点对点通讯（队列）生产者 队列 消费者 关系
消息中间件队列的作用：生产者：主要向队列发送消息。消费者：主要从队列中获取消息步骤：1、生产者向队列进行发送消息，如果消费者不在，队列就会将消息缓存2、消费者向队列中获取到消息之后，消费成功之后，该消息队列直接被清除掉。(不清除就会产生重复消费问题)
生产者向队列中生产高并发流量，消费者会不会挂掉？答：不会，因为队列会缓存消息。为什么MQ能够解决高并发问题？答：不会立马处理那么多的消息，队列会进行缓存，进行排队。
队列可以做持久化的
三、消息中间件通讯方式
JMS：java发送消息，客户端与服务器进行通讯的方式，可以理解成java的消息中间件消息模型：1、点对点通讯方式：流程：生产者 队列 消费者2、发布订阅流程：生产者 主题 消费者第一种通讯方式：点对点通讯方式： 一对一 异步通讯，生产者生产的消息只允许有一个消费者进行消费。
第二种通讯方式：发布订阅 一个生产者 可以多个消费者 一对多
四、消息中间件应用场景
消息中间件应用场景：1、点对点进行通讯：一对一。没有使用消息中间件流程：
使用消息中间件后流程：
点对点通讯，程序效率提高
2、发布订阅：一对多。应用：推送，但是一般不会做，都会使用第三方推送。如:极光推送、融云、环信
五、ActivityMQ 安装
MQ种类：RabbitMQ、Redis（也可以做发布订阅）、ZeroMQ、Kafka、RocketMQ、OnesMQ(阿里收费）kafka主要是存放日志的，
六、使用ActivityMQ发送消息
导入依赖：
<!-- 引入activityMQ依赖包 --><dependencies><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-core</artifactId><version>5.7.0</version></dependency></dependencies>
生产者代码：ProducerTest.java
package com.leeue.activemq.test01;import javax.jms.Connection;import javax.jms.DeliveryMode;import javax.jms.Destination;import javax.jms.JMSException;import javax.jms.MessageProducer;import javax.jms.Queue;import javax.jms.Session;import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnection;import org.apache.activemq.ActiveMQConnectionFactory;/*** @classDesc: 功能描述:(这是一个生产者 使用activityMQ)* @author:<a href="leeue@">李月</a>* @Version:v1.0* @createTime:9月21日 下午5:51:45*/public class ProducerTest {public static void main(String[] args) throws JMSException {//1、获取mq连接工程ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD,"tcp://127.0.0.1:61616"); // 这个是MQ真正的后台通讯地址。//2、创建连接Connection createConnection = activeMQConnectionFactory.createConnection();//3、启动连接createConnection.start();//4、创建会话工厂 设置成默认的Session session = createConnection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);//5.创建队列Destination destination = session.createQueue("leeueMsg");//6. 得到生产者MessageProducer producer = session.createProducer(destination);//7.开始生产 这里设置成不持久化producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);for(int i = 0 ; i < 5; i++) {senMsg(session, producer, "我是生产者:"+i);}System.out.println("生产者发送消息完毕...");}public static void senMsg(Session session, MessageProducer producer,String msg) throws JMSException {//1.创建文本消息TextMessage textMessage = session.createTextMessage("Hello MQ"+msg);//2.发送消息producer.send(textMessage);}}
当生产者发送消息的时候，没有消费者，那么生产者发送的消息就会存放在队列里面去图：
消费者代码：
package com.leeue.activemq.test01;import javax.jms.Connection;import javax.jms.DeliveryMode;import javax.jms.Destination;import javax.jms.JMSException;import javax.jms.Message;import javax.jms.MessageConsumer;import javax.jms.MessageProducer;import javax.jms.Session;import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnection;import org.apache.activemq.ActiveMQConnectionFactory;/*** @classDesc: 功能描述:(这是个消费者)* @author:<a href="leeue@">李月</a>* @Version:v1.0* @createTime:9月22日 下午12:25:29*/public class Consumer001 {public static void main(String[] args) throws JMSException {// 1、获取mq连接工程ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616"); // 这个是MQ真正的后台通讯地址。// 2、创建连接Connection createConnection = activeMQConnectionFactory.createConnection();// 3、启动连接createConnection.start();// 4、创建会话工厂 设置成默认的Session session = createConnection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);// 5.创建队列Destination destination = session.createQueue("leeueMsg");// 6. 创建一个消费者MessageConsumer consumer = session.createConsumer(destination);// 7.建立一个while循环一直监听消息while (true) {// 监听消息 因为刚刚消费者传送的报文就是 TextMessage格式的，所以这边强转就行了.TextMessage textMessage = (TextMessage) consumer.receive();if (textMessage != null) {System.out.println("消费者获取到消息:" + textMessage.getText());} else {break;}}System.out.println("消费者消费完毕...");}}
七、ActivityMQ的消息订阅
1、主题代码ProducerTest02.java
package com.leeue.activemq.test02;import javax.jms.Connection;import javax.jms.DeliveryMode;import javax.jms.Destination;import javax.jms.JMSException;import javax.jms.MessageProducer;import javax.jms.Queue;import javax.jms.Session;import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnection;import org.apache.activemq.ActiveMQConnectionFactory;/*** @classDesc: 功能描述:(这是一个主题 使用activityMQ)* @author:<a href="leeue@">李月</a>* @Version:v1.0* @createTime:9月21日 下午5:51:45*/public class ProducerTest02 {public static void main(String[] args) throws JMSException {//1、获取mq连接工程ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD,"tcp://127.0.0.1:61616"); // 这个是MQ真正的后台通讯地址。//2、创建连接Connection createConnection = activeMQConnectionFactory.createConnection();//3、启动连接createConnection.start();//4、创建会话工厂 设置成默认的Session session = createConnection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);//5. 得到生产者MessageProducer producer = session.createProducer(null);//6.开始生产 这里设置成不持久化producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);for(int i = 0 ; i < 5; i++) {senMsg(session, producer, "我是生产者主题:"+i);}System.out.println("生产者发送消息完毕 主题...");}public static void senMsg(Session session, MessageProducer producer,String msg) throws JMSException {//1.创建文本消息TextMessage textMessage = session.createTextMessage("Hello MQ"+msg);//2.创建一个通道Destination destination = session.createTopic("leeueTopic");//3.指定主题，然后发送这个消息producer.send(destination,textMessage);}}
2、订阅者代码 Consumer002.java 这个跟消费者没什么区别，、就是获取订阅主题的消息而已
package com.leeue.activemq.test02;import javax.jms.Connection;import javax.jms.DeliveryMode;import javax.jms.Destination;import javax.jms.JMSException;import javax.jms.Message;import javax.jms.MessageConsumer;import javax.jms.MessageProducer;import javax.jms.Session;import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnection;import org.apache.activemq.ActiveMQConnectionFactory;/*** @classDesc: 功能描述:(这是个消费者)* @author:<a href="leeue@">李月</a>* @Version:v1.0* @createTime:9月22日 下午12:25:29*/public class Consumer002 {public static void main(String[] args) throws JMSException {// 1、获取mq连接工程ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616"); // 这个是MQ真正的后台通讯地址。// 2、创建连接Connection createConnection = activeMQConnectionFactory.createConnection();// 3、启动连接createConnection.start();// 4、创建会话工厂 设置成默认的Session session = createConnection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);// 5.创建一个主题Destination destination = session.createTopic("leeueTopic");// 6. 创建一个消费者MessageConsumer consumer = session.createConsumer(destination);// 7.建立一个while循环一直监听消息while (true) {// 监听消息 因为刚刚消费者传送的报文就是 TextMessage格式的，所以这边强转就行了.TextMessage textMessage = (TextMessage) consumer.receive();if (textMessage != null) {System.out.println("消费者获取主题发布的消息:" + textMessage.getText());} else {break;}}System.out.println("消费者消费完毕...");}}
主题和订阅者是实时在通信的，并且是一个主题可以对应多个消费者。
Java架构学习（三十三）ActivityMQ基础消息中间件概述异步与同步MQ作用MQ件通讯方式MQ应用场景ActivityMQ安装使用ActivityMQ的案例主题和订阅
