消息队列学习 --ActiveMQ概念了解

Posted 2022-08-02 躬匠

上一章节我们学习了RabbitMQ中间件的相关知识，今天我们就来接着学习另一个中间件ActiveMQ。

一、简介

AcitiveMQ是由Apache出品的开源消息中间件，可提供高效、可扩展、稳定、安全的企业级消息通信。其具有以下特性：

• 支持Java、php、C、C++、Python等多种语言的客户端
• 提供了消息组通信、消息优先级、延迟接收消息、消息持久化等高级特性
• 支持分布式事务消息、事务
• 连接方式多样化，例如TCP、UDP、SSL、多播等
• 提供了技术和语言中立的REST API接口
• 支持以Ajax方式调用ActiveMQ
• 为高性能集群、客户端-服务器、点对点通信场景而设计

一、基本概念

1、消息传送模型

• 点对点模型。使用对嘞作为消息通信载体，满足生产者和消费者模式。一条消息只能被一个消费者所使用，未被消费的消息将在队列中保留直到被消费或者超时
• 发布/订阅模型。使用主题作为消息通信载体，类似于广播模式，发布者发布一条消息，该消息通过主题传递给所有的订阅者。

2、基本组件

• Broker（消息代理）：消息队列服务器实体，接受客户端连接，提供消息通信的核心服务
• Producer（生产者）：负责生产消息并发送给Broker
• Consumer（消费者）：从Broker接收消息并进行业务逻辑处理
• Topic（主题）：发布/订阅模式下生产者向Topic发送消息，并由Broker分发给不同的订阅者，实现消息的广播
• Quene（队列）：点对点模式下生产者向队列发送消息，消费者订阅特定队列接收消息并进行业务逻辑处理
• Message（消息）：按照不同的通信协议定义的固定格式编码的数据包

3、消息存储

ActiveMQ支持消息的持久化和非持久化分发方式。

非持久化分发指的是Broker保证尽最大可能分发消息，但是并不会持久化存储消息；而持久化存储指的是在发送前先将消息持久化存储。

非持久化消息常用于发送通知或者实时数据中。当你比较看重性能，即使丢失一些消息也不会影响业务正常运作时，可选择非持久化消息；

持久化消息被发送到Broker之后，如果当前消息的消费者并没有运行，则该消息继续存在。只有在消息被消费者处理而且被ACK确认之后，才会从Broker中删除消息。

消息可以被持久化存储到内存、文件或者数据库中。

三、应用场景

前面的一篇文章我们概述过消息队列的应用场景，有异步处理、应用解耦、分布式事务、日志收集等等。那RabbitMQ的应用场景又具体有哪些呢？

1、消息推送

前面我们学习RabbitMQ时介绍过消息推送，ActiveMQ的消息推送与RabbitMQ相同，这里就不再介绍了。

2、异步处理/流量削峰

这个就不用说了，所有的消息队列都具有的功能。

3、分布式事务

目前常见的解决方案有两阶段提交（2PC）、事务补偿（TCC)、本地事务表 + 消息队列、MQ事务消息。这里我们聚焦一下ActiveMQ的事务表 + 消息队列的方案。

场景：用户注册成功之后向用户发放积分。其中，用户模块和积分模块是两个独立的服务，数据库也是独立的。

假如用户模块服务的数据库为DB1、积分服务的数据库为DB2。

解决方案：

在DB1、DB2两数据库上分别建立一个事务表，事务表里存储事务的类型、事务处理状态以及事件内容；

将添加用户记录和添加事务记录放到一个本地事务中，其中，事务表中的事件类型为新建用户，状态为新建；

用户服务服务器会存在一个定时任务，定时扫描事务表中事件类型为新建用户的事件，找出来之后发送一条消息到消息队列服务器；只有在消息发送成功之后才会修改事务表中状态为已发布，否者就会一直发送消息，直到发送成功。

消息发送到消息队列服务器之后，消息队列服务器会向积分服务推送消息（这个过程可能会存在消息发送失败的情况，由消息队列服务器保证消息的可靠投递），消息的内容就是事务表中存储的内容。

积分服务接收到这个消息之后会在本地的事务表中插入一条事务记录，事务的状态为已发布，内容就是传递的消息，类型为新建用户。

积分服务也有一个定时任务定时查询事务表中状态为已发布的记录，并读取其中的内容。之后在本地事务中为用户增加积分并修改事务表的状态为已处理。

这种事务表 + 消息队列的方式思想上借鉴了2PC，但是又与其有所不同：2PC是同步事务，而事务表+消息队列将事务变成了异步执行，提高了吞吐量。

四、其他

消息积压

消息积压的场景有很多，如果发送的消息没有得到及时回复或者消费者的消费速度跟不上生产者的生产速度，都会导致持久化消息不断积压而得不到释放，从而阻塞队列。

对于这种情况，可以通过配置消息的过期时间和死信队列来预防。

1.过期时间

默认情况下，ActiveMQ的消息永远不会过期。但是可以为某个消息或者所有的消息设置过期时间，如果发送消息后，在消息过期时间到时消息还没有被发送到目的地，则该消息将被清除（也有可能会被放到死信队列中）。

2.死信队列

死信队列是用来保存处理失败或者过期的消息的。

另外，对于ActiveMQ，也需要设置消息应答模式来显示的告知Broker消息已经处理成功，否者会导致Broker一直在等待，直到超时或者重试。

消息发送优化

1.异步发送

生产者可以选择同步或者异步的方式向消息队列服务器发送数据。

当消费者的消费速度跟不上生产者的生产速度时，使用同步模式会造成消息生产者的堵塞，从而影响生产者的吞吐量。

所以，一般情况下，如果消费消息的速度比较快，则可以使用同步模式；如果消费消息的速度比较慢，则建议使用异步的消息传递模式。

2.生产者流量控制

在消息积压并超过限制大小的情况下，要对消息生产者进行限流。

可以在超过限制大小时让消息生产者进入等待状态或者直接抛出异常，可以通过配置指定。

当消息被再次投递的次数超过配置的最大次数之后（默认为6次），Broker便会将该消息放入死信队列中。

消息消费优化

1.慢速消费者

消息消费速度慢于消息生产速度的消费者都称为慢速消费者。

慢速消费者会给Broker带来潜在危险，因为这会导致消息堆积，大量消息驻留在内存中。一旦内存耗尽将会导致消息数据不断从文件读取到内存，然后又被交换到文件，从而消耗磁盘I/O；更严重的是会拖累消息生产者，导致生产者一侧阻塞而减慢生产者发送消息的速度，从而牵连原本快速消费者也无法获取到充足的消息消费，降低吞吐量。

所以，一定要根据应用场景对慢速消费者以及潜在的危险做一些容错；

目前用的比较多的是消息丢弃：当等待消息达配置的上限时，有新增的消息到来时将根据不同的粗略丢弃旧的消息。主要有以下三种策略：

• 丢弃最旧的消息
• 丢弃最旧且优先级最低的消息
• genuine自定义属性来丢弃消息

消息持久化

ActiveMQ的消息持久化和RabbitMQ的持久化思想上很类似：当生产者发出消息之后，Broker首先将消息存储到文件、数据库、内存中，然后再将消息发送给消费者，发送成功之后则会将消息从相应的存储中删除，若失败则会尝试继续发送。

这样Broker在启动时会自动先检查指定位置的存储，如发现有发送失败的消息，则会继续发送。