目录
一. 前言
二. JMS 规范
2.1. 基本概念
2.2. JMS 体系结构
三. ActiveMQ 使用
3.1. ActiveMQ Classic 和 ActiveMQ Artemis
3.2. Queue 模式(P2P)
3.3. Topic 模式(Pub/Sub)
3.4. 持久订阅
3.5. 消息传递的可靠性
3.5.1. 事务型会话与非事务型会话
3.5.2. 持久化与非持久化消息的存储策略
3.6. 消息发送策略
3.7. Failover 参数配置
四. 原理解析
4.1. 生产消息原理
4.2. 消费消息原理
4.3. 消息确认及消息重发
五. 基本优化
六. 总结
一. 前言
一开始消息中间件的厂商繁多,且各个厂商之间没有统一的规范,这就导致了各消息中间件非常难以整合协作,因此,后来陆续出现了如 JMS 和 AMQP 这样的消息队列规范,提供了统一的标准,而 ActiveMQ 就是完全遵循 JMS 规范开发的消息队列。
二. JMS 规范
2.1. 基本概念
什么是 JMS(Java Message Service)规范?JMS 是一个基于 Java 平台面向消息中间件(MOM)的 API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。在设计JMS 时,设计师就计划能够结合现有消息队列的优点,如:
- 不同的消息传送模式或域,例如点对点消息传送和发布/订阅消息传送;
- 支持同步和异步消息;
- 支持可靠性消息的传输;
- 支持常见的消息格式,如:文本、字节、流、对象等。
2.2. JMS 体系结构
上图是对 JMS 的总体介绍,下面对其中各个对象分别进行说明:
1. ConnectionFactory:连接工厂,一般设为单例模式,一旦创建,就一直运行在应用容器内,客户端使用连接工厂创建一个JMS连接。
2. Connection:JMS 连接,表示 JMS 客户端和服务器端之间的一个活动的连接。
3. Session:JMS 会话,表示 JMS 客户端与 JMS 服务器端之间的会话状态。JMS 会话建立在JMS 连接上,表示客户与服务器之间的一个会话线程。
4. Destination:消息管道,从生产端流向客户端,包括队列(PTP),主题(Pub/Sub)。
5. Message Producer 和 Message Consumer:生产者和消费者对象由 Session 对象创建,用于发送和接收消息。
6. Message:JMS 消息由以下几部分组成:消息头,属性,消息体。
- 消息头(Header):JMS 消息头包含了许多字段,它们是消息发送后由 JMS 提供者或消息发送者产生,用来表示消息、设置优先权和失效时间等等,并且为消息确定路由 Routing。
- 属性(Property):由消息发送者产生,用来添加删除消息头以外的附加信息。
- 消息体(Body):由消息发送者产生,JMS 中定义了5种消息体:ByteMessage、MapMessage、ObjectMessage、StreamMessage 和 TextMessage。
三. ActiveMQ 使用
ActiveMQ 支持 P2P(点对点)传输和 Pub/Sub 模型,这两种传递方式的本质区别就是消息是否可重复消费。比如微信私聊和群聊,私聊就是 P2P,除了私聊的双方其它人无法再获取消息,而群聊就相当于 Pub/Sub 模式,即群成员都订阅了该群的消息。
在讲消息传输之前,我们先看看 ActiveMQ 官网发布的两种版本的区别。
3.1. ActiveMQ Classic 和 ActiveMQ Artemis
ActiveMQ Classic 官网地址:https://activemq.apache.org/components/classic/
ActiveMQ Artemis 官网地址:https://activemq.apache.org/components/artemis/
实际上 ActiveMQ Classic 原来就叫 ActiveMQ,是 Apache 开发的基于 JMS 1.1 的消息服务器,目前稳定版本号是 6.x,而 ActiveMQ Artemis 是由 RedHat 捐赠的 HornetQ 服务器代码的基础上开发的,目前稳定版本号是 2.x。
和 ActiveMQ Classic 相比,Artemis 版的代码与 Classic 完全不同,并且,它支持JMS 2.0,使用基于 Netty 的异步 IO,大大提升了性能。针对数据的持久层 ActiveMQ Artemis 还能够支持 JDBC。
整体来说,ActiveMQ 通常指的是 ActiveMQ Classic,为了简化去掉了 Classic。ActiveMQ Artemis 应该是作为下一个版本来候选的,支持的协议更新。2 套 ActiveMQ 的代码是不一样的。如果用不到什么太多的消息策略高级需求,使用 ActiveMQ Classic 就好。因为这 2 个消息服务器的代码完全不一样,导致如果你使用 Spring 的话,使用的包的代码也不一样,ActiveMQ Artems 的调用代码更加简单,界面更好看。
3.2. Queue 模式(P2P)
先创建一个 Producer 生产消息:
public static void main(String[] args) {ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.0.106:61616");Connection connection = null;try {// 创建并开启连接connection = factory.createConnection();connection.start();// 创建会话,设置是否为事务型会话以及消息签收方式Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);// 创建发送队列Destination destination = session.createQueue("queue");// 创建消息发送者MessageProducer producer = session.createProducer(destination);// 创建消息并设置消息内容TextMessage textMessage = session.createTextMessage();textMessage.setText("Hello");// 发送消息producer.send(textMessage);session.commit();session.close();} catch (JMSException e) {e.printStackTrace();} finally {try {connection.close();} catch (JMSException e) {e.printStackTrace();}}
}
上面代码注释写的很清楚了,可以看到是完全符合 JMS 的体系结构的,首先创建一个连接工厂,并通过连接工厂创建连接,然后通过连接创建会话(在创建会话时可以指定是否为事务型会话以及设置消息的签收方式,相关概念在后面会详细讲解),之后再为本次会话创建管道,即传输队列(这里可以指定是创建队列(Queue)还是还是主题(Topic)),最后创建消息对象发送到管道提交即完成本次会话的消息生产。
接下来看看消费者如何消费消息:
public static void main(String[] args) {ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.0.106:61616");Connection connection = null;try {connection = factory.createConnection();connection.start();Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);// 创建接收队列Destination destination = session.createQueue("queue");// 创建消息消费者MessageConsumer consumer = session.createConsumer(destination);// 接收消息TextMessage message = (TextMessage) consumer.receive();System.out.println(message.getText());session.commit();session.close();} catch (JMSException e) {e.printStackTrace();} finally {try {connection.close();} catch (JMSException e) {e.printStackTrace();}}
}
整个流程和生产者流程基本是一样的,只不过消费者不再需要自己生产消息,而是从消息队列中获取,这里是通过 receive() 方法获取的,该方法相当于是客户端主动从队列中“拉”消息,并且在消息队列为空时会阻塞等待消息传入。
另外还有一种队列“推”送的方式,通过监听器实现:
public static void main(String[] args) {ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.0.106:61616");Connection connection = null;try {connection = factory.createConnection();connection.start();Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);Destination destination = session.createQueue("queue");MessageConsumer consumer = session.createConsumer(destination);// 使用监听器监听队列MessageListener listener = new MessageListener() {@Overridepublic void onMessage(Message message) {try {System.out.println(((TextMessage) message).getText());} catch (JMSException e) {e.printStackTrace();}}};while (true) {consumer.setMessageListener(listener);session.commit();}} catch (JMSException e) {e.printStackTrace();} finally {try {connection.close();} catch (JMSException e) {e.printStackTrace();}}
}
需要注意的是 listener 不会阻塞等待,当消息到达时会主动调用 onMessage() 方法,但它的生命周期和方法的生命周期是相同的,需要像上面一样死循环监听,同时 receive 和 listener 是互斥的,即同时只能使用其中一种方式来获取消息。
3.3. Topic 模式(Pub/Sub)
相对于 P2P,发布订阅模式就是可以有多个消费者监听同一个队列,并可重复消费同一个消息,整个代码实现流程和上面的是一样的,只是将 Destination destination = session.createQueue(“queue”); 改为 Destination destination = session.createTopic(“topic”); 即可。
这里需要思考一个问题,消费者能够订阅到哪个时间段的消息呢?是所有的消息还是自消费者注册监听之后的呢?很显然,肯定是只能获取到注册监听之后的消息。但是,若是消费者中途怠机再恢复,怠机过程中产生的消息能否接收到呢?AcitveMQ 是支持获取怠机过程中的消息的,即持久订阅功能。
3.4. 持久订阅
什么是持久订阅?举个例子,相当于你在微博点击关注某个博主,无论你是否在线,博主发送的消息你都是可以获取到的,持久订阅就类似这样,在创建好连接后首先设置一个自身的身份标识clientId,这个 id 是唯一的:
connection.setClientID("lhzm");
然后通过下面 API 创建消费者即可创建持久订阅:
MessageConsumer consumer = session.createDurableSubscriber((Topic) destination, "lhzm");
需要注意持久订阅只有 Pub/Sub 模式下才支持。
3.5. 消息传递的可靠性
在学习了基础的使用后,我们应该考虑一个问题,消息队列该如何保证消息传递的可靠性呢?即如何保证生产的消息正确被消费者签收或者被生产者销毁?这就牵涉到事务型会话和非事务型会话,JMS Session 接口提供了 commit 和 rollback 方法。
事务提交意味着生产的所有消息被发送,消费的所有消息被确认;事务回滚意味着生产的所有消息被销毁,消费的所有消息被恢复并重新提交,除非它们已经过期。 事务型的会话总是牵涉到事务处理中,commit 或 rollback 方法一旦被调用,一个事务就结束了,而另一个事务被开始;关闭事务性会话将回滚其中的事务。
3.5.1. 事务型会话与非事务型会话
JMS 在创建 session 会话时通过第一个参数指定是否为事务型会话:
Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
当为事务型会话时,调用 commit 方法前消息并不会真正的投递到消息中间件中去,而在调用commit 后消息会自动确认,需要保证发送端和接收端都是事务型会话。
当为非事务型会话时,相当于生产者逐个投递到消息中间件,但是消息的确认取决于消费者如何设置 ACK_MODE,即创建会话时的第二个参数,该参数有4个选项:
- SESSION_TRANSACTED:当为事务型会话时的默认选项,若不是事务型会话设置该参数会抛出异常。
- AUTO_ACKNOWLEDGE:当消费者成功的从 receive() 方法返回的时候,或者从MessageListenner.onMessage() 方法成功返回的时候,会话自动确认客户收到消息。
- CLIENT_ACKNOWLEDGE:消费者通过调用 Message 的 acknowledge() 方法确认消息。需要注意该模式下何时调用 acknowledge() 方法,那么在调用该方法之前收到的消息都会一起被确认,而在此之后收到的消息不会被确认。比如,发送10条消息,消费者在收到第5条消息时调用 acknowledge() 方法,那么前5条都会被确认。
- DUPS_OK_ACKNOWLEDGE:消息延迟批量确认,消息生产者在消费者没有确认消息时会重新发送消息。该模式可优化消费者确认消息的性能,但可能会导致消费者收到重复消息。
需要注意第一个是和事务绑定,后面三个都是针对消费端的,即消息中间件需要接收到消费者的ack 才会认为消息被正确处理。
3.5.2. 持久化与非持久化消息的存储策略
消息队列为保证高效,消息首先肯定是存储在内存中的,那么一旦消息队列怠机或者消息过多超出内存,消息就会面临丢失的风险,所以需要有相关的手段来保证。
正常情况下,非持久化消息是存储在内存中的,能够存储的最大消息数据在 /conf/activemq.xml 文件中的 systemUsage 节点可配置:
<systemUsage><systemUsage><memoryUsage><memoryUsage percentOfJvmHeap="70" /></memoryUsage><storeUsage><storeUsage limit="100 gb"/></storeUsage><tempUsage><tempUsage limit="50 gb"/></tempUsage></systemUsage>
</systemUsage>
memoryUsage 是设置整个 ActiveMQ 节点的“可用内存限制”。这个值不能超过 ActiveMQ 本身设置的最大内存大小。其中的 percentOfJvmHeap 属性表示百分比。
storeUsage 是设置整个 ActiveMQ 节点,用于存储“持久化消息”的“可用磁盘空间”。
tempUsage 是设置临时文件大小。一旦 ActiveMQ 服务节点存储的消息达到了 memoryUsage 的限制,非持久化消息就会被转储到 temp store 区域,虽然我们说过非持久化消息不进行持久化存储,但是 ActiveMQ 为了防止数据洪峰出现时,非持久化消息大量堆积致使内存耗尽的情况出现,还是会将非持久化消息写入到磁盘的临时区域——temp store。
从上文我们可以了解到 ActiveMQ的 存储策略,但是还有个问题,持久化消息是通过什么介质存储的呢?主要有以下 5 种:
1. KahaDB:默认的存储方式。在 data/kahadb 这个目录下,会生成四个文件:
- db-*.log:存储消息内容。新的数据以 APPEND 的方式追加到日志文件末尾。属于顺序写入,因此消息存储是比较快的。默认是32M,达到阀值会自动递增。
- db.data:它是消息的索引文件,本质上是 B-Tree(B树),使用 B-Tree 作为索引指向 db-*.log 里面存储的消息。
- db.redo:用来进行消息恢复。
- lock 文件锁:表示当前获得 kahadb 读写权限的 broker。
2. JDBC存储:需要配置 JDBC 连接以及引入相应的 jar。会在数据库创建三张表:
- ACTIVEMQ_MSGS:消息表,Queue 和 Topic 都存在这个表中;
- ACTIVEMQ_ACKS:存储持久订阅的信息和最后一个持久订阅接收的消息 ID。
- ACTIVEMQ_LOCKS:锁表,用来确保某一时刻,只能有一个 ActiveMQ broker 实例来访问数据库。
3. Memory存储:即内存。
4. LevelDB存储:性能优于 KahaDB,但官方不推荐使用。
5. JDBC Message store with ActiveMQ Journal:这种方式克服了 JDBC Store 的不足,JDBC每次消息过来,都需要去写库和读库。ActiveMQ Journal,使用高速缓存写入技术,大大提高了性能。
详细配置方式参照官方文档。
3.6. 消息发送策略
ActiveMQ 支持同步、异步两种发送模式将消息发送到消息中间件上。
同步发送过程中,发送者发送一条消息会阻塞直到消息中间件反馈一个确认消息,表示消息已经被消息中间件处理。这个机制提供了消息的安全性保障,但是由于是阻塞的操作,会影响到客户端消息发送的性能。
异步发送的过程中,发送者不需要等待 broker 提供反馈,所以性能相对较高。但是可能会出现消息丢失的情况。所以使用异步发送的前提是在某些情况下允许出现数据丢失的情况。
默认情况下,非持久化消息是异步发送的,持久化消息并且是在非事务模式下是同步发送的。但是在开启事务的情况下,消息都是异步发送。由于异步发送的效率会比同步发送性能更高,所以在发送持久化消息的时候,尽量去开启事务会话。除了持久化消息和非持久化消息的同步和异步特性以外,我们还可以通过以下几种方式来设置异步发送:
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.0.106:61616?jms.useAsyncSend=true");
((ActiveMQConnectionFactory) connectionFactory).setUseAsyncSend(true);
((ActiveMQConnection) connection).setUseAsyncSend(true);
3.7. Failover 参数配置
默认的情况下,这种协议用于随机的去选择一个链接去连接,如果连接失败了,那么会连接到其他的 Broker 上。默认配置定义了延迟重新连接,意味着传输将会在 10s 后自动去重新连接可用的broker。当然所以有的重新连接参数都可以根据应用的需要而配置的。
选项名 | 默认值 | 说明 |
---|---|---|
backup | false | 提前初始化一个未使用的链接,以便进行快速的失败转移,默认false |
initialReconnectDelay | 10 | 在第一次尝试重新连接之前等待的时间长度(毫秒),默认10 |
maxCacheSize | 131072 | 当trackMessage启动时,缓存的最大子接,默认为127*1024bytes |
maxReconnectAttempts | -1 | 0 | 默认1|0,自5.6版本开始,-1为默认值,代表不限重试次数,0标识从不重试(只尝试连接一次,并不重连),5.6以前的版本,0为默认值,代表不重试,如果设置大于0的数,则代表最大重试次数。 |
maxReconnectDelay | 30000 | 最长重连时间间隔(毫秒),默认30000 |
nested.* | null | 来自ActiveMQ 5.9:将应用于列表中每个URI的通用URI选项。 |
randomize | true | 使用随机连接,以达到负载均衡的目的,默认true |
reconnectDelayExponent | 2.0 | 在指数后退尝试过程中使用的指数。 |
reconnectSupported | true | 确定客户端是否应通过重新连接来响应 broker ConnectionControl 事件(请参阅:rebalanceClusterClients)。 |
startupMaxReconnectAttempts | -1 | 初始化时的最大重试次数,一旦连接上,将使用maxReconnectAttempts的配置,默认0 |
timeout | -1 | 从ActiveMQ 5.3开始:在不中断重新连接过程的情况下,设置发送操作的超时(以毫秒为单位)。 |
trackMessages | false | 设置是否缓存(故障发生时)尚未传送完成的消息,当broker一旦重新连接成功,便将这些缓存中的消息刷新到新连接的代理中,使得消息可以在broker切换前后顺利传送。默认false |
updateURIsSupported | true | 设定是否可以动态修改 broker uri(自5.4版本开始),默认true |
updateURIsURL | null | 从ActiveMQ 5.4:指向文本文件的URL(或本地文件的路径),该文本文件包含一个逗号分隔的URI列表,用于在失败时重新连接。 |
useExponentialBackOff | true | 重连时间间隔是否以指数形式增长,默认true |
warnAfterReconnectAttempts | 10 | 从ActiveMQ 5.10:值>0指定在记录警告之前重新连接尝试的次数。记录的警告表示当前没有连接,但正在尝试重新连接。值<=0将禁用有关重新连接尝试的警告的日志记录。 |
四. 原理解析
ActiveMQ 的上手非常简单,但仅仅只是会用肯定不行,只有了解其原理,才能对特定的场景做出优化和设计,而要了解其原理,只有通过分析其源码才能完全了解。接下来针对生产消息、消费消息和消息重发机制的流程做一个概括性总结。
4.1. 生产消息原理
生产消息流程
上面就是整个发消息的流程图,当生产者调用 send 发送消息时,首先会判断producerWindowSize 是否还有空间,若没有了就阻塞等待空间;反之则继续判断是否是异步发送消息,如果是同步,则直接通过底层传输协议传输消息,并阻塞等待 response 结果;如果是异步发送,同样通过底层传输协议传输消息,但不再需要阻塞等待 response,同时会去增加 producerWindowSize 的值。
什么是 producerWindowSize?这个配置主要用来约束异步发送时 producer 端允许积压(未 ack)的消息大小。当发送消息时,首先会判断 producerWindowSize 是否还有剩余空间,如果没有就阻塞等待空间释放,即等待 broker(可以就当作是消息队列中间件)确认消息;如果有空间,就放入到该空间下,等待 broker 处理。可以通过以下两种方式配置:
- 在连接 url 中设置,对所有 producer 都有效:tcp://localhost:61616?jms.producerWindowSize=1048576
- 在 destination 名称中设置,仅对使用该 destination 的 producer 有效,并且优先级更高:test-queue?producer.windowSize=1048576
4.2. 消费消息原理
消费消息流程
消费者在通过 receive() 消费消息时,并不是直接去 broker 上获取的消息,而是从本地的unconsumerMessage 队列中获取,而该队列则是每次批量从 broker 上拉取消息,每次拉取的数量就是由 prefetchSize 控制的。当队列中没有消息时,就会阻塞等待获取消息;反之则依次从unconsumerMessage 队列中取出消息消费,并将应答放到 delivered 队列返回给 broker,消费消息和 ack 是异步的。下面我们来看看消息的确认过程。
4.3. 消息确认及消息重发
看到上面这张图,可能会比较懵,没关系,我们首先来了解一下 ACK_MODE 和 ACK_TYPE,ACK_MODE 在上文已经讲过了,但仅仅是消费端确认了还不够,还需要让 broker 知道消息是否正常消费,因此在确认消息后消费者还会根据处理结果返回不同的 ACK_TYPE 给 broker,ACK_TYPE 一共有以下 6 种:
- DELIVERED_ACK_TYPE = 0 消息"已接收",但尚未处理结束。
- POSION_ACK_TYPE = 1 消息"错误",通常表示“抛弃”此消息,比如消息重发多次后,都无法正确处理时,消息将会被删除或者加入 DLQ(死信队列)。
- STANDARD_ACK_TYPE = 2 “标准"类型,通常表示为消息“处理成功”,broker 端可以删除消息了。
- REDELIVERED_ACK_TYPE = 3 消息需“重发”,比如 consumer 处理消息时抛出了异常,broker 稍后会重新发送此消息。
- INDIVIDUAL_ACK_TYPE = 4 表示无论在任何 ACK_MODE 下只确认“单条消息”。
- UNMATCHED_ACK_TYPE = 5 在 Topic 中,如果一条消息在转发给“订阅者”时,发现此消息不符合 Selector 过滤条件,那么此消息将不会转发给订阅者,消息将会被存储引擎删除(相当于在 Broker 上确认了消息)。
- EXPIRED_ACK_TYPE = 6 消息已过期。
清楚了 ACK_TYPE 所对应的意思后,再看这张图就很明了了。首先从 unconsumerMessage 队列中取出消息并处理,若消费消息出现异常失败,消费者就会返回 REDELIVERED_ACK_TYPE 给broker,broker 就会重发该条消息,当超过次数限制消费者就会返回 POSION_ACK_TYPE 告诉broker 该条消息是有毒的,broker 根据配置将该条消息抛弃或是加入死信队列中(该队列可以被重新消费);若消费消息成功未出现异常,就会将 ack message 添加到 delivered 队列中,消费该队列的消息时,会进行一系列判断并根据结果返回不同的 ACK_TYPE。
刚刚我们提到消息消费失败会导致消息重发,那究竟在哪些情况下会被重发呢?主要有以下几种情况:
- 在事务型会话中,若是没有调用 session.commit() 提交确认消息或者调用 session.rollback() 方法。
- 在非事务性会话中,ACK 模式为 CLIENT_ACKNOWLEDGE 的情况下,没有调用 acknowledge() 或者调用了 recover() 方法。
- 处理消息时发生异常。
这就是整个消息的确认和重发原理。
五. 基本优化
在上文我们提到过 prefetchSize 配置,该配置表示消费者每次从队列中获取消息的条数,该配置为 0 时表示消费者通过 pull 方式从 broker 获取消息,另外不同类型的队列具有不同的默认值:
- 持久化队列和非持久化队列的默认值为 1000;
- 持久化 topic 默认值为 100;
- 非持久化 topic 的默认值为 Short.MAX_VALUE-1。
但是仅仅只有批量获取肯定是不够的,因为从上文我们知道,消息还有一个确认过程,如果还是单个单个的确认,那这个批量获取就没有什么意义了(除了第一次是批量获取消息,后面都是单个单个的获取消息),所以 ActiveMQ 还提供了 optimizeAcknowledge 配置,该参数为 true 时,消费者会延迟确认(默认是 ack 了 0.65*prefetchSize 个消息后才确认)。该配置可以直接在连接url中配置(其中 optimizeAcknowledgeTimeOut 是表示超过该时间也会自动确认):
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory( "tcp://192.168.0.106:61616?jms.optimizeAcknowledge=true&jms.optimizeAcknowledgeTimeOut=10000");
因此,这两者协同配合才能起到优化的作用。另外,需要注意的是,如果消费端的消费速度比较高,通过这两者组合能大大提升消费者的性能。如果消费者的消费性能本身就比较慢,设置比较大的 prefetchSize 反而不能有效的达到提升消费性能的目的,因为过大的 prefetchSize 会导致某一消费端积压消息,而其它的消费端却“无所事事”。同时,该方案需要消费端能够容忍重复消息,因为当消息还未确认时消费者就怠机了,那么 broker 就会将该消息重发给其它消费者,导致消息重复。
六. 总结
通过以上学习,我们能看出 ActiveMQ 是非常简单易上手的,但它有以下缺点:
- 持久化消息存储需要建立索引,因此吞吐量低,不适合 TPS 要求高的业务。
- 不支持消息分片功能,只能自己实现。