JMS规范、ActiveMQ Broker和ActiveMQ传输协议

Java实现ActiveMQ通讯（构建过程）

编写pom.xml配置文件

<!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-all --><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-all</artifactId><version>5.15.11</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.xbean/xbean-spring --><dependency><groupId>org.apache.xbean</groupId><artifactId>xbean-spring</artifactId><version>4.15</version></dependency>

JMS编码总体架构

队列

在点对点的消息传递域中，目的地被称为队列（queue）

点对点消息传递域的特点如下：

每个消息只能有一个消费者，类似于1对1的关系。好比个人快递自己领自己的。
消息的生产者和消费者之间没有时间上的相关性。无论消费者在生产者发送消息的时候是否处于运行状态，消费者都可以提取消息。好比我们的发送短信，发送者发送后不见得接收者会即收即看。
消息被消费后队列中不会再存储，所以消费者不会消费到已经被消费掉的消息。

演示案例

（1）消息生产者

package com.htzw.activemq;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class JMSProducer {public static final String ACTIVEMQ_URL = "tcp://192.168.67.130:61616";public static final String QUEUE_NAME = "queue01";public static void main(String[] args) throws JMSException {// 1.创建连接工厂，按照给定的url地址，采用默认用户名和密码ActiveMQConnectionFactory 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 <= 3; i++) {//7.创建消息TextMessage textMessage = session.createTextMessage("msg-----" + i);//8.通过messageProducer发送给MQ队列messageProducer.send(textMessage);}//9.关闭相关资源messageProducer.close();session.close();connection.close();System.out.println("-----消息发布到MQ完成");}
}

（2）控制台说明

取值	说明
Number Of Pending Messages	等待消费的消息，这个是未出队列的数量，公式=总接收数-总出队列数
Number Of Consumers	消费者数量，消费者端的消费者数量
Messages Enqueued	进队消息数，进队列的总消息量，包括出队列的。这个数只增不减
Messages Dequeued	出队消息数，可以理解为是消费者消费掉的数量

当有一个消息进入这个队列时，等待消费的消息是1，进入队列的消息是1。

当消息消费后，等待消费的消息是0，进入队列的消息是1，出队列的消息是1。

当再来一条消息时，等待消费的消息是1，进入队列的消息就是2。

（3）消息消费者

方式一：阻塞式消费者

package com.demo.activemq.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/*** 简单消息消费者*/
public class JmsConsumer {public static final String ACTIVEMQ_URL = "tcp://192.168.67.130:61616";public static final String QUEUE_NAME = "queue01";public static void main(String[] args) throws JMSException {//1.创建连接工厂，按照给定的URL，采用默认的用户名密码ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2.通过连接工厂,获得connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3.创建会话session//两个参数transacted=事务,acknowledgeMode=确认模式(签收)Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4.创建目的地(具体是队列queue还是主题topic)Queue queue = session.createQueue(QUEUE_NAME);//5.创建消息的消费者,指定消费哪一个队列里面的消息MessageConsumer messageConsumer = session.createConsumer(queue);//循环获取while (true) {//6.通过消费者调用方法获取队列里面的消息(发送的消息是什么类型,接收的时候就强转成什么类型)TextMessage textMessage = (TextMessage) messageConsumer.receive();if (textMessage != null) {System.out.println("****消费者接收到的消息:  " + textMessage.getText());}else {break;}}//7.关闭资源messageConsumer.close();session.close();connection.close();}
}

方式二：异步监听式消费者

package com.demo.activemq.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
import java.io.IOException;
/*** 监听模式下的消费者*/
public class JmsConsumer2 {public static final String ACTIVEMQ_URL = "tcp://192.168.67.130:61616";public static final String QUEUE_NAME = "queue01";public static void main(String[] args) throws JMSException, IOException {//1.创建连接工厂，按照给定的URL，采用默认的用户名密码ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2.通过连接工厂,获得connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3.创建会话session//两个参数transacted=事务,acknowledgeMode=确认模式(签收)Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4.创建目的地(具体是队列queue还是主题topic)Queue queue = session.createQueue(QUEUE_NAME);//5.创建消息的消费者,指定消费哪一个队列里面的消息MessageConsumer messageConsumer = session.createConsumer(queue);//6.通过监听的方式消费消息/*异步非阻塞式方式监听器(onMessage)订阅者或消费者通过创建的消费者对象,给消费者注册消息监听器setMessageListener,当消息有消息的时候,系统会自动调用MessageListener类的onMessage方法我们只需要在onMessage方法内判断消息类型即可获取消息*/messageConsumer.setMessageListener(new MessageListener() {@Overridepublic void onMessage(Message message) {if (message != null && message instanceof TextMessage) {//7.把message转换成消息发送前的类型并获取消息内容TextMessage textMessage = (TextMessage) message;try {System.out.println("****消费者接收到的消息:  " + textMessage.getText());} catch (JMSException e) {e.printStackTrace();}}}});System.out.println("执行了39行");//保证控制台不关闭,阻止程序关闭System.in.read();//关闭资源messageConsumer.close();session.close();connection.close();}
}

（4）小结

JMS开发的基本步骤

JMS开发的主要步骤如下：

创建一个connection factory
通过connection factory来创建JMS connection
启动JMS connection
通过JMS connection创建JMS session
创建JMS destination（目的地队列/主题）
创建JMS producer或者创建JMS consume并设置destination
创建JMS consumer或者注册一个JMS message listener
发送(send)或者接收(receive)JMS message
关闭所有JMS资源

两种消费方式

消费方式	说明
同步阻塞方式（receive）	订阅者或接收者只用MessageConsumer的receive()方法来接收消息，receive方法在能接收到消息之前（或超时之前）将一直阻塞
异步非阻塞方式（监听器onMessage()）	订阅者或接收者通过MessageConsumer的setMessageListener(MessageListener listener)注册一个消息监听器，当消息到达之后，系统会自动调用监听器MessageListener的onMessage(Message message)方法

主题

在发布订阅消息传递域中,目的地被称为主题(topic)

发布/订阅消息传递域的特点如下：

生产者将消息发布到topic中，每个消息可以有多个消费者，属于1：N的关系；
生产者和消费者之间有时间上的相关性。订阅某一个主题的消费者只能消费自它订阅之后发布的消息。
生产者生产时，topic不保存消息它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息，所以，一般先启动消费者再启动生产者。

JMS规范允许客户创建持久订阅，这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。一句话，好比我们的微信公众号订阅

演示案例

重点说明：先启动订阅者，再启动生产者，不然发送的消息是废消息

（1）发布主题生产者


package com.demo.activemq.topic;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class JmsProducer_Topic {public static final String ACTIVEMQ_URL = "tcp://192.168.67.130:61616";public static final String TOPIC_NAME = "topic01";public static void main(String[] args) throws JMSException {//1.创建连接工厂，按照给定的URL，采用默认的用户名密码ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2.通过连接工厂,获得connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3.创建会话session//两个参数transacted=事务,acknowledgeMode=确认模式(签收)Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4.创建目的地(具体是队列queue还是主题topic)Topic topic = session.createTopic(TOPIC_NAME);//5.创建消息的生产者MessageProducer messageProducer = session.createProducer(topic);//6.通过使用消息生产者,生产三条消息,发送到MQ的队列里面for (int i = 0; i < 3; i++) {//7.通过session创建消息TextMessage textMessage = session.createTextMessage("TOPIC_NAME---" + i);//8.使用指定好目的地的消息生产者发送消息messageProducer.send(textMessage);}//9.关闭资源messageProducer.close();session.close();connection.close();System.out.println("****TOPIC_NAME消息发布到MQ完成");}
}

（2）订阅主题消费者


package com.demo.activemq.topic;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;
import java.io.IOException;public class JmsConsumer_Topic {public static final String ACTIVEMQ_URL = "tcp://192.168.67.130:61616";public static final String TOPIC_NAME = "topic01";public static void main(String[] args) throws JMSException, IOException {System.out.println("我是1号消费者");//1.创建连接工厂，按照给定的URL，采用默认的用户名密码ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2.通过连接工厂,获得connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3.创建会话session//两个参数transacted=事务,acknowledgeMode=确认模式(签收)Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4.创建目的地(具体是队列queue还是主题topic)Topic topic = session.createTopic(TOPIC_NAME);//5.创建消息的消费者MessageConsumer messageConsumer = session.createConsumer(topic);//5.创建消息的消费者,指定消费哪一个队列里面的消息messageConsumer.setMessageListener(message -> {if (message instanceof TextMessage){try {String text = ((TextMessage) message).getText();System.out.println(text);} catch (JMSException e) {e.printStackTrace();}}});System.in.read();}
}

小结

两种模式的效果

两种模式的对比

对比维度	Topic主题模式	Queue队列模式
工作模式	“订阅-发布”模式，如果当前没有订阅者，消息将会被丢弃，如果有多个订阅者，那么这些订阅者都会收到消息	“负载均衡”模式，如果当前没有消费者，消息也不会丢弃，如果有多个消费者，那么一条消息也只会发送给其中一个消费者，并且要求消费者ACK消息
有无状态	无状态	Queue数据默认会在MQ服务器上已文件形式保存，比如ActiveMQ一般保存在$AMQ_HOME\data\kr-store\data下面，也可以配置成DB存储
传递完整性	如果没有订阅者，消息会被丢弃	消息不会被丢弃
处理效率	由于消息要按照订阅者的数量进行复制，所以处理性能会随着订阅者的增加而明显降低，并且还要结合不同消费协议自身的性能差异	由于一条消息只发送一个消费者，所以就算消费者再多，性能也不会有明显降低。当然不同消费协议的具体性能也是有差异的

JMS规范和落地产品

是什么

JavaEE

JavaEE是一套使用Java进行企业级应用开发的大家一致遵循的13个核心规范工业标准。JavaEE平台提供了一个基于组件的方法来加快设计，开发。装配及部署企业应用程序。

名称	说明
JDBC（Java Databease）	数据库连接
JNDI（Java Naming and Directory Interfaces）	Java的命令和目录接口
EJB	Enterprise JavaBean
RMI（Remote Method Invoke）	远程方法调用
Java IDL（Interface Description Language）/CORBA（Common Object Broker Architecture）	接口定义语言/共用对象请求代理程序体系结构
JSP	Java Server Page
Servlet
XML（Extensible Markup Language）	可标记白标记语言
JMS（Java Message Service）	Java消息服务
JTA（Java Transaction API）	Java事务API
JTS（Java Transaction Service）	Java事务服务
JavaMail
JAF（JavaBean Activation Framework）

JMS

Java Message Service(Java消息服务是JavaEE中的一个技术。

什么是Java消息服务？Java消息服务指的是两个应用程序之间进行异步通信的API，它为标准协议和消息服务提供了一组通用接口，包括创建、发送、读取消息等，用于支持Java应用程序开发。

在JavaEE中，当两个应用程序使用JMS进行通信时，它们之间不是直接相连的，而是通过一个共同的消息收发服务组件关联起来以达到解耦/异步削峰的效果。

术语

ActiveMQ符合JMS规范。需要了解有关该规范的一些术语：

术语	说明
Provider/MessageProvider	生产者
Consumer/MessageConsumer	消费者
PTP	Point To Point，点对点通信消息模型
Pub/Sub	Publish/Subscribe，发布订阅消息模型
Queue	队列，目标类型之一，和PTP结合
Topic	主题，目标类型之一，和Pub/Sub结合
ConnectionFactory	连接工厂，JMS用它创建连接
Connnection	JMS Client到JMS Provider的连接
Destination	消息目的地，由Session创建
Session	会话，由Connection创建，实质上就是发送、接受消息的一个线程，因此生产者、消费者都是Session创建的

MQ中间件的其他落地产品

特性	ActiveMQ	RabbitMQ	Kafka	RocketMQ
Producter+consumer	支持	支持	支持	支持
Publish+Subscribe	支持	支持	支持	支持
Request+Reply	支持	支持	——	支持
API完备性	高	高	高	低（静态配置）
多语言支持	支持，Java优先	语言无关	支持，Java优先	支持
单机吞吐量	万级	万级	十万级	单机万级
消息延迟	——	微秒级	毫秒级	——
可用性	高（主从）	高（主从）	非常高（分布式）	高
消息丢失	——	低	理论上不会丢失	——
消息重复	——	可控制	理论上会有重复	——
文档完备性	高	高	高	中
提供快速入门	有	有	有	无
首次部署难度	——	低	中	高

JMS的组成结构和特点

JMS Provider

实现JMS接口和规范的消息中间件，也就是我们说的MQ服务器

JMS Producer

消息生产者，创建和发送JMS消息的客户端应用

JMS Consumer

消息消费者，接收和处理JMS消息的客户端应用

JMS Message

（1）消息头

JMSDestination
- 消息发送的目的地，主要是指Queue和Topic
JMSDeliveryMode
- 持久模式和非持久模式
- 一条持久性的消息：应该被传送“一次仅仅一次”，这就意味着如果JMS提供者出现故障，该消息并不会丢失，它会在服务器恢复之后再次传递
- 一条非持久的消息：最多会传递一次，这意味着服务器出现故障，该消息将会永远丢失
JMSExpiration
- 可以设置消息在一定时间后过期，默认是永不过期
- 消息过期时间，等于Destination的send方法中的timeToLive值加上发送时刻的GMT时间值
- 如果timeToLive值等于0，则JMSExpiration被设为0，表示该消息永不过期
- 如果发送后，在消息过期时间之后还没有被发送到目的地，则该消息被清除
JMSPriority
- 消息优先级，从0-9十个级别，0-4是普通消息，5-9是加急消息
- JMS不要求MQ严格按照这十个优先级发送消息但必须保证加急消息要先于普通消息到达。默认是4级
JMSMessageID
- 唯一标识每个消息的标识由MQ产生

（2）消息属性

封装具体的消息数据
5种消息格式
- TxtMessage：普通字符串消息，包含一个String
- MapMessage：一个Map类型的消息，key为Strng类型，而值为Java基本类型
- BytesMessage：二进制数组消息，包含一个byte[]
- StreamMessage：Java数据流消息，用标准流操作来顺序填充和读取
- ObjectMessage：对象消息，包含一个可序列化的Java对象
发送和接收的消息体类型必须一致对应

（3）消息体

如果需要除消息字段以外的值，那么可以使用消息属性
识别/去重/重点标注等操作非常有用的方法

JMS的可靠性

PERSISTENT:持久性

参数设置说明

非持久
- messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT)
- 非持久化：当服务器宕机，消息不存在。
持久
- messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT)
- 持久化：当服务器宕机，消息依然存在。
Queue默认是持久

持久的Queue

MessageProducer messageProducer = session.createProducer(queue);
//设置通过session创建出来的生产者生产的Queue消息为持久性
messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);

持久化消息：这是队列的默认传递模式，此模式保证这些消息只被传送一次和成功使用一次。对于这些消息，可靠性是优先考虑的因素。
可靠性的另一个重要方面是确保持久性消息传送至目标后，消息服务在向消费者传送它们之前不会丢失这些消息。

持久的Topic

先启动定阅消费者再启动定阅生产者

持久的发布主题生产者

package com.demo.activemq.persist;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;/*** 持久化Topic生产者*/
public class JmsProducer_Topic_Persist {private static final String ACTIVEMQ_URL = "tcp://192.168.67.130:61616";private static final String ACTIVEMQ_TOPIC_NAME = "Topic-Persist";public static void main(String[] args) throws JMSException {//1.创建连接工厂，按照给定的URL，采用默认的用户名密码ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2.通过连接工厂,持久化的topic必须在生产者创建并设置持久化完成后调用startConnection connection = activeMQConnectionFactory.createConnection();//3.创建会话session//两个参数transacted=事务,acknowledgeMode=确认模式(签收)Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4.创建目的地(具体是队列queue还是主题topic)Topic topic = session.createTopic(ACTIVEMQ_TOPIC_NAME);//5.创建消息的生产者MessageProducer messageProducer = session.createProducer(topic);//6.设置生产者生产持久化的TopicmessageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);//7.启动连接connection.start();//8.通过使用持久化Topic消息生产者,生产三条消息,发送到MQ的队列里面for (int i = 0; i < 3; i++) {//7.通过session创建消息TextMessage textMessage = session.createTextMessage("msg-persist" + i);//8.使用指定好目的地的消息生产者发送消息messageProducer.send(textMessage);}//9.关闭资源messageProducer.close();session.close();connection.close();System.out.println("****TOPIC_NAME消息发布到MQ完成");}
}

情况一：订阅者在线

情况二：订阅者不在线

持久的订阅主题消费者

package com.demo.activemq.persist;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;
import java.io.IOException;/*** 持久化Topic消费者*/
public class Jms_Topic_Consumer_Persist {private static final String ACTIVEMQ_URL = "tcp://192.168.67.130:61616";private static final String ACTIVEMQ_TOPIC_NAME = "Topic-Persist";public static void main(String[] args) throws JMSException, IOException {System.out.println("我是3号消费者王五");//1.创建连接工厂，按照给定的URL，采用默认的用户名密码ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2.通过连接工厂,获得connection,设置connectionIDConnection connection = activeMQConnectionFactory.createConnection();connection.setClientID("王五");//3.创建会话session//两个参数transacted=事务,acknowledgeMode=确认模式(签收)Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4.创建目的地(具体是队列queue还是主题topic)Topic topic = session.createTopic(ACTIVEMQ_TOPIC_NAME);//5.通过session创建持久化订阅TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic, "我是王五");//6.启动连接connection.start();//7.接收消息topicSubscriber.setMessageListener(message -> {if (message instanceof TextMessage) {TextMessage textMessage = (TextMessage) message;try {System.out.println("收到的持久化订阅消息: " + textMessage.getText());} catch (JMSException e) {e.printStackTrace();}}});/*** 一定要先运行一次消费者,类似于像MQ注册,我订阅了这个主题* 然后再运行主题生产者* 无论消费着是否在线,都会接收到,在线的立即接收到,不在线的等下次上线把没接收到的接收*/}
}

情况一：订阅者在线

情况二：订阅者不在线

控制台

Transaction：事务

producer提交时的事务

false
- 只要执行send，就进入到队列中
- 关闭事务，那第2个签收参数的设置需要有效
true
- 先执行send再执行commit，消息才被真正提交到队列中
- 消息需要批量提交，需要缓冲处理

事务偏生产者/签收偏消费者

代码—生产者


package com.demo.activemq.tracation;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class Jms_TX_Producer {private static final String ACTIVEMQ_URL = "tcp://192.168.67.130:61616";private static final String ACTIVEMQ_QUEUE_NAME = "Queue-TX";public static void main(String[] args) throws JMSException {//1.创建连接工厂，按照给定的URL，采用默认的用户名密码ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2.通过连接工厂,获得connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3.创建会话session//两个参数transacted=事务,acknowledgeMode=确认模式(签收)//开启事务需要commitSession session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);//4.创建目的地(具体是队列queue还是主题topic)Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);//5.创建消息的生产者,并设置不持久化消息MessageProducer producer = session.createProducer(queue);//6.通过使用消息生产者,生产三条消息,发送到MQ的队列里面try {for (int i = 0; i < 3; i++) {TextMessage textMessage = session.createTextMessage("tx msg--" + i);producer.send(textMessage);}//7.提交事务session.commit();System.out.println("消息发送完成");} catch (Exception e) {System.out.println("出现异常,消息回滚");session.rollback();} finally {//8.关闭资源producer.close();session.close();connection.close();} }
}

代码—消费者

package com.demo.activemq.tracation;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class Jms_TX_Consumer {private static final String ACTIVEMQ_URL = "tcp://192.168.67.130:61616";private static final String ACTIVEMQ_QUEUE_NAME = "Queue-TX";public static void main(String[] args) throws JMSException {//1.创建连接工厂，按照给定的URL，采用默认的用户名密码ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);//2.通过连接工厂,获得connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();//3.创建会话session//两个参数transacted=事务,acknowledgeMode=确认模式(签收)//消费者开启了事务就必须手动提交，不然会重复消费消息Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);//4.创建目的地(具体是队列queue还是主题topic)Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);//5.创建消息的消费者,指定消费哪一个队列里面的消息MessageConsumer messageConsumer = session.createConsumer(queue);//6.通过监听的方式消费消息messageConsumer.setMessageListener(new MessageListener() {int a = 0;@Overridepublic void onMessage(Message message) {if (message instanceof TextMessage) {try {if (a == 2) {System.out.println(1 / 0);}TextMessage textMessage = (TextMessage) message;System.out.println("***消费者接收到的消息:   " + textMessage.getText());session.commit();a = a + 1;} catch (Exception e) {System.out.println("出现异常，消费失败，放弃消费");try {session.rollback();a=0;} catch (JMSException ex) {ex.printStackTrace();}}}}});//7.关闭资源}
}

Acknowledge：签收

非事务

自动签收（默认方式）
- Session.AUTO_ACKNOWLEDGE
手动签收
- Session.CLIENT_ACKNOWLEDGE
- 客户端调用acknowledge方法手动签收
允许重复消息

事务

生产事务开启，只有commit后才能将全部消息变为已消费

消息生产者

package com.demo.activemq.acknowledge;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class Jms_Transaction_AUTOACK_Producer {private static final String ACTIVEMQ_URL = "tcp://192.168.67.130:61616";private static final String ACTIVEMQ_QUEUE_NAME = "Queue-ACK-NoTransaction";public static void main(String[] args) throws JMSException {ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);Connection connection = activeMQConnectionFactory.createConnection();connection.start();Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);MessageProducer producer = session.createProducer(queue);for (int i = 0; i < 3; i++) {TextMessage textMessage = session.createTextMessage("Transaction_AUTOACK-msg:   " + i);producer.send(textMessage);}session.commit();System.out.println("发送完成");producer.close();session.close();connection.close();}
}

消息消费者


package com.demo.activemq.acknowledge;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class Jms_Transaction_CLIENTACK_Consumer {private static final String ACTIVEMQ_URL = "tcp://192.168.67.130:61616";private static final String ACTIVEMQ_QUEUE_NAME = "Queue-ACK-Transaction";public static void main(String[] args) throws JMSException {ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);Connection connection = activeMQConnectionFactory.createConnection();connection.start();//消费者设置了手动签收,就必须自己签收,向服务器发送我已经收到消息了//开启事务如果不提交,就算手动签收,也是无效的Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);MessageConsumer messageConsumer = session.createConsumer(queue);messageConsumer.setMessageListener(new MessageListener() {@Overridepublic void onMessage(Message message) {if (message instanceof TextMessage) {TextMessage textMessage = (TextMessage) message;try {textMessage.acknowledge();System.out.println(textMessage.getText());} catch (JMSException e) {e.printStackTrace();}}}});}
}

由于消费者开启了事务，没有提交事务(就算手动签收也没用)，服务器认为，消费者没有收到消息

签收和事务的关系

JMS的点对点总结

点对点模型是基于队列的，生产者发送消息到队列，消费者从队列接收消息，队列的存在使得消息的异步传输成为可能。
如果在Session关闭时有部分消息被收到但还没有被签收（acknowledge），那当消费者下次连接到相同的队列时，这些消息还会被再次接收
队列可以长久的保存消息直到消费者收到消息。消费者不需要因为担心消息会丢失而时刻和队列保持激活的链接状态，充分体现了异步传输模式的优势

JMS的发布订阅总结

JMS Pub/Sub 模型定义了如何向一个内容节点发布和订阅消息，这些节点被称作topic
主题可以被认为是消息的传输中介，发布者（publisher）发布消息到主题，订阅者（subscribe）从主题订阅消息
主题使得消息订阅者和消息发布者保持互相独立不需要解除即可保证消息的传送

非持久订阅

非持久订阅只有当客户端处于激活状态，也就是和MQ保持连接状态才能收发到某个主题的消息。
如果消费者处于离线状态，生产者发送的主题消息将会丢失作废，消费者永远不会收到。
先订阅注册才能接受到发布，只给订阅者发布消息

持久订阅

客户端首先向MQ注册一个自己的身份ID识别号，当这个客户端处于离线时，生产者会为这个ID保存所有发送到主题的消息，当客户再次连接到MQ的时候，会根据消费者的ID得到所有当自己处于离线时发送到主题的消息
当持久订阅状态下，不能恢复或重新派送一个未签收的消息
持久订阅才能恢复或重新派送一个未签收的消息。

当所有的消息必须被接收，则用持久订阅。当消息丢失能够被容忍，则用非持久订阅

ActiveMQ的Broker

概述

Broker相当于一个ActiveMQ服务器实例
Broker其实就是实现了用代码的形式启动ActiveMQ将MQ嵌入到Java代码中，以便随时用随时启动，
在用的时候再去启动这样能节省了资源，也保证了可用性。

嵌入式Broker

POM.XML

<dependencies><!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-all --><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-all</artifactId><version>5.15.11</version></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>2.10.1</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.xbean/xbean-spring --><dependency><groupId>org.apache.xbean</groupId><artifactId>xbean-spring</artifactId><version>4.15</version></dependency>     <dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>1.2.17</version>
</dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.10</version></dependency>
</dependencies>

Embedoker

import org.apache.activemq.broker.BrokerService;public class EmbedBroker {public static void main(String[] args) throws Exception {//ActiveMQ也支持在vm中通信基于嵌入的brokerBrokerService brokerService = new BrokerService();brokerService.setPopulateJMSXUserID(true);brokerService.addConnector("tcp://127.0.0.1:61616");brokerService.start();}
}

队列验证

和Linux上的ActiveMQ是一样的,Broker相当于一个Mini版本的ActiveMQ

ActiveMQ的传输协议

概述

ActiveMQ支持的client-broker通讯协议有：TVP、NIO、UDP、SSL、Http(s)、VM。

其中配置Transport Connector的文件在ActiveMQ安装目录的conf/activemq.xml中的标签之内。见下图实际配置：

在上文给出的配置信息中，URI描述信息的头部都是采用协议名称：例如

描述amqp协议的监听端口时，采用的URI描述格式为“amqp://······”；
描述Stomp协议的监听端口时，采用URI描述格式为“stomp://······”；
唯独在进行openwire协议描述时，URI头却采用的“tcp://······”。这是因为ActiveMQ中默认的消息协议就是openwire

传输协议的种类

Transmission Control Protocol（TCP）默认协议

这是默认的Broker配置，TCP的Client监听端口61616
在网络传输数据前，必须要先序列化数据，消息是通过一个叫wire protocol的来序列化成字节流。
TCP连接的URI形式如：tcp://HostName:port?key=value&key=value，后面的参数是可选的。
TCP传输的的优点：
- TCP协议传输可靠性高，稳定性强
- 高效率：字节流方式传递，效率很高
- 有效性、可用性：应用广泛，支持任何平台
- 关于Transport协议的可选配置参数可以参考官网http://activemq.apache.org/configuring-version-5-transports.html

New I/O API Protocol(NIO)

NIO协议和TCP协议类似，但NIO更侧重于底层的访问操作。它允许开发人员对同一资源可有更多的client调用和服务器端有更多的负载。
适合使用NIO协议的场景：
- 可能有大量的Client去连接到Broker上，一般情况下，大量的Client去连接Broker是被操作系统的线程所限制的。因此，NIO的实现比TCP需要更少的线程去运行，所以建议使用NIO协议。
- 可能对于Broker有一个很迟钝的网络传输，NIO比TCP提供更好的性能。
NIO连接的URI形式：nio://hostname:port?key=value&key=value
关于Transport协议的可选配置参数可以参考官网http://activemq.apache.org/configuring-version-5-transports.html

AMQP协议

Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同开发语言等条件限制。

Stomp协议

STOP，Streaming Text Orientation Message Protocol，是流文本定向消息协议，是一种为MOM(Message Oriented Middleware，面向消息中间件)设计的简单文本协议。

Secure Sockets Layer Protocol(SSL)

MQTT协议

MQTT(Message Queuing Telemetry Transport，消息队列遥测传输)是IBM开发的一个即时通讯协议，有可能成为物联网的重要组成部分。

该协议支持所有平台，几乎可以把所有联网物品和外部连接起来，被用来当作传感器和致动器(比如通过Twitter让房屋联网)的通信协议。

WS协议(websocket)

ActiveMQ支持的网络协议

协议	描述
TCP	默认的协议，性能相对可以
NIO	基于TCP协议之上的，进行了扩展和优化，具有更好的扩展性
UDP	性能比TCP更好，但是不具有可靠性
SSL	安全连接
HTTP(S)	基于HTTP或者HTTPS
VM	VM本身不是协议，当客户端和代理在同一个JAVA虚拟机中运行时，他们之间需要通信，但不想占用网络通信，而是直接通信，可以使用该方式