RocketMQ

生产者（Producer）

主题（Topic）

队列（MessageQueue）

消息（Message）

消费者分组（ConsumerGroup）

消费者（Consumer）

订阅关系（Subscription）

同步的RPC远程调用

基于中间件代理的异步通信方式

系统拓扑简单

上下游耦合性弱

容量削峰填谷

点对点模型

发布订阅模型

总结

定义数据的分类隔离： 建议将不同业务类型的数据拆分到不同的主题中管理，通过主题实现存储的隔离性和订阅隔离性

定义数据的身份和权限： 消息本身是匿名无身份的，同一分类的消息使用相同的主题来做身份识别和权限管理。

用于标识主题，主题名称集群内全局唯一

一个主题内包含一个或多个队列，消息实际存储在主题的各队列内

普通消息

顺序消息

定时/延时消息

事务消息

消息类型必须一致：发送的消息的类型，必须和目标主题定义的消息类型一致

主题类型必须单一：每个主题只支持一种消息类型，不允许将多种类型的消息发送到同一个主题中

6（110）：可读写

4（100）：只读

2（010）：只写

0（000）：不可读写

是什么：消费者消费消息时，Apache RocketMQ 向消费者客户端投递消息的顺序

根据不同的消费场景，Apache RocketMQ 提供顺序投递和并发投递两种方式

默认是并发投递

最大重试次数（默认16次）

重试间隔（频率递减的方式）

使用PushConsumer类型的消费者消费消息时，消费者客户端必须设置消费监听器

如上图所示，以在线的电商交易场景为例，上游订单系统将用户下单支付这一业务事件封装成独立的普通消息并发送至Apache RocketMQ服务端，下游按需从服务端订阅消息并按照本地消费逻辑处理下游任务。每个消息之间都是相互独立的，且不需要产生关联。

如上图所示，以离线的日志收集场景为例，通过埋点组件收集前端应用的相关操作日志，并转发到 Apache RocketMQ 。每条消息都是一段日志数据，Apache RocketMQ 不做任何处理，只需要将日志数据可靠投递到下游的存储系统和分析系统即可，后续功能由后端应用完成。

延申思考：超店中的PV表页面浏览路径，也是通过埋点收集前端操作日志并发给后端，后端发给MQ异步处理

初始化

待消费

消费中

消费提交

消息删除

在分布式定时调度场景下，需要实现各类精度的定时任务，例如每天5点执行文件清理，每隔2分钟触发一次消息推送等需求。传统基于数据库的定时调度方案在分布式场景下，性能不高，实现复杂。基于 Apache RocketMQ 的定时消息可以封装出多种类型的定时触发器。

以电商交易场景为例，订单下单后暂未支付，此时不可以直接关闭订单，而是需要等待一段时间后才能关闭订单。使用 Apache RocketMQ 定时消息可以实现超时任务的检查触发。

Apache RocketMQ 定时消息设置的定时时间是一个预期触发的系统时间戳，延时时间也需要转换成当前系统时间后的某一个时间戳，而不是一段延时时长。

定时时长最大值默认为24小时，不支持自定义修改

初始化

定时中：消息被发送到服务端，和普通消息不同的是，服务端不会直接构建消息索引，而是会将定时消息单独存储在定时存储系统中，等待定时时刻到达。

待消费

消费中

消费提交

消息删除

定时消息的实现逻辑需要先经过定时存储等待触发，定时时间到达后才会被投递给消费者。因此，如果将大量定时消息的定时时间设置为同一时刻，则到达该时刻后会有大量消息同时需要被处理，会造成系统压力过大，导致消息分发延迟，影响定时精度。

以证券、股票交易撮合场景为例，对于出价相同的交易单，坚持按照先出价先交易的原则，下游处理订单的系统需要严格按照出价顺序来处理订单

以数据库变更增量同步场景为例，上游源端数据库按需执行增删改操作，将二进制操作日志作为消息，通过 Apache RocketMQ 传输到下游搜索系统，下游系统按顺序还原消息数据，实现状态数据按序刷新。如果是普通消息则可能会导致状态混乱，和预期操作结果不符，基于顺序消息可以实现下游状态和上游操作结果一致。

生产顺序性

消费顺序性

与普通消息的生命周期一致

备注

以电商交易场景为例，用户支付订单这一核心操作的同时会涉及到下游物流发货、积分变更、购物车状态清空等多个子系统的变更

为了保证上述四个分支的执行结果一致性，典型方案是基于XA协议的分布式事务系统来实现。将四个调用分支封装成包含四个独立事务分支的大事务。基于XA分布式事务的方案可以满足业务处理结果的正确性，但最大的缺点是多分支环境下资源锁定范围大，并发度低，随着下游分支的增加，系统性能会越来越差。（概括为：有一个事务协调者，多个事务参与者，由协调者统一调度commit或者rollback。协调者有著名的alibaba的seata）

将上述基于XA事务的方案进行简化，将订单系统变更作为本地事务，剩下的系统变更作为普通消息的下游来执行，事务分支简化成普通消息+订单表事务，充分利用消息异步化的能力缩短链路，提高并发度。该方案中消息下游分支和订单系统变更的主分支很容易出现不一致的现象。

如：
消息发送成功，订单没有执行成功，需要回滚整个事务。

订单执行成功，消息没有发送成功，需要额外补偿才能发现不一致。

消息发送超时未知，此时无法判断需要回滚订单还是提交订单变更。

上述普通消息方案中，普通消息和订单事务无法保证一致的原因，本质上是由于普通消息无法像单机数据库事务一样，具备提交、回滚和统一协调的能力。
而基于Apache RocketMQ实现的分布式事务消息功能，在普通消息基础上，支持二阶段的提交能力。将二阶段提交和本地事务绑定，实现全局提交结果的一致性。

Apache RocketMQ 事务消息保证本地主分支事务和下游消息发送事务的一致性，但不保证消息消费结果和上游事务的一致性。因此需要下游业务分支自行保证消息正确处理，建议消费端做好消费重试，如果有短暂失败可以利用重试机制保证最终处理成功。

Apache RocketMQ 事务消息的生命周期存在超时机制，即半事务消息被生产者发送服务端后，如果在指定时间内服务端无法确认提交或者回滚状态，则消息默认会被回滚。事务超时时间，默认值是4小时

一般出现消息回查时事务正在处理的原因为：事务执行较慢，消息回查太快。解决方案如下：

将第一次事务回查时间设置较大一些，但可能导致依赖回查的事务提交延迟较大。

程序能正确识别正在进行中的事务。

在客户端SDK中内置请求重试逻辑

同步发送和异步发送模式均支持消息发送重试

生产者在初始化时设置消息发送最大重试次数，生产者发送失败则进行重试

除服务端返回系统流控错误场景，其他触发条件触发重试后，均会立即进行重试，无等待间隔

若由于服务端返回流控错误触发重试，系统会按照指数退避策略进行延迟重试（第一次失败重试前后需等待多久，默认值：1秒）

链路耗时阻塞评估。（重试可能会对链路造成阻塞延时，如果后续操作对实时性要求比较高，需要调整合理超时时间以及重试次数）

最终异常兜底（重试不一定最终发送成功，需要业务方做兜底保证数据一致性）

消息重复问题（出现假超时，其实是发送成功并到达服务端，只需拿到响应报文太慢，又重试发送了一次）

存储压力大（消费位点为最大位点 || 需回溯到指定时刻前开始消费）

服务端请求任务排队溢出（消息堆积）

客户端会收到

监控系统容量，保证底层资源充足

如果因为突发原因触发消息流控，且客户端内置的重试流程执行失败，则建议业务方将请求调用临时替换到其他系统进行应急处理。

消息获取

消息处理

消息状态提交

若您的业务场景发生变更，或您当前使用的消费者类型不适合当前业务，您可以选择在 PushConsumer 和SimpleConsumer 之间变更消费者类型。变更消费者类型不影响当前Apache RocketMQ 资源的使用和业务处理。

在实际使用场景中，PullConsumer 仅推荐在流处理框架中集成使用，大多数消息收发场景使用 PushConsumer 和 SimpleConsumer 就可以满足需求。

生产环境中相同的 ConsumerGroup 下严禁混用 PullConsumer 和其他两种消费者，否则会导致消息消费异常。

PushConsumer 消费消息时，若消息处理逻辑出现预期之外的阻塞导致消息处理一直无法执行成功，SDK会按照消费超时处理强制提交消费失败结果，并按照消费重试逻辑进行处理。消息超时，请参见PushConsumer消费重试策略。

出现消费超时情况时，SDK虽然提交消费失败结果，但是当前消费线程可能仍然无法响应中断，还会继续处理消息。

在PushConsumer类型中，消息的实时处理能力是基于SDK内部的典型Reactor线程模型（底层使用队列）实现的。如下图所示，SDK内置了一个长轮询线程，先将消息异步拉取到SDK内置的缓存队列中，再分别提交到消费线程中，触发监听器执行本地消费逻辑。

客户端SDK严格按照监听器的返回结果判断消息是否消费成功，并做可靠性重试。

所有消息必须以同步方式进行消费处理，并在监听器接口结束时返回调用结果，不允许再做异步化分发

不允许使用以下方式处理消息，否则 Apache RocketMQ 无法保证消息的可靠性

如果消费者分组设置了顺序消费模式，则PushConsumer在触发消费监听器时，严格遵循消息的先后顺序。

PushConsumer严格限制了消息同步处理及每条消息的处理超时时间，适用于以下场景

try {
    // SimpleConsumer 需要主动获取消息，并处理。
    List<MessageView> messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
    messageViewList.forEach(messageView -> {
        System.out.println(messageView);
        // 消费处理完成后，需要主动调用 ACK 提交消费结果。
        try {
            simpleConsumer.ack(messageView);
        } catch (ClientException e) {
            logger.error("Failed to ack message, messageId={}", messageView.getMessageId(), e);
        }
    });
} catch (ClientException e) {
    // 如果遇到系统流控等原因造成拉取失败，需要重新发起获取消息请求。
    logger.error("Failed to receive message", e);
}

客户端SDK如果处理消息成功则调用AckMessage接口；如果处理失败只需要不回复ACK响应，即可在定义的消费不可见时间到达后触发消费重试流程

基于 Apache RocketMQ 顺序消息的定义，SimpleConsumer在处理顺序消息时，会按照消息存储的先后顺序获取消息。即需要保持顺序的一组消息中，如果前面的消息未处理完成，则无法获取到后面的消息

消息处理时长不可控

需要异步化、批量消费等高级定制场景

需要自定义消费速率

对于PushConsumer消费类型，需要严格控制消息的消费耗时，尽量避免出现消息处理超时导致消息重复。如果业务经常会出现一些预期外的长时间耗时的消息，建议使用SimpleConsumer，并设置好消费不可见时间。

将符合条件的消息投递给消费者，而不是将匹配到的消息过滤掉

子主题

消息类型是否一致：普通消息、顺序消息需要按topic拆分

业务域是相同

消息的量级和重要性是否一致

PushConsumer和SimpleConsumer默认负载策略

PullConsumer默认负载策略

是什么：每条消息在队列中都有一个唯一的Long类型坐标，这个坐标被定义为消息位点

最早一条消息的位点为最小消息位点（MinOffset）；最新一条消息的位点为最大消息位点（MaxOffset）

消息队列逻辑上是无限存储，但由于服务端物理节点的存储空间有限， Apache RocketMQ 会滚动删除队列中存储最早的消息

消息的最小消费位点和最大消费位点会一直递增变化

Apache RocketMQ 通过消费位点管理消息的消费进度

RocketMQ会维护一份记录，记录某个消费者组消费某个队列的最新消息的位点

当消费者客户端离线，又再次重新上线时，会严格按照服务端保存的消费进度继续处理消息。
如果服务端保存的历史位点信息已过期被删除，此时消费位点向前移动至服务端存储的最小位点

消费位点的保存和恢复是基于 Apache RocketMQ 服务端的存储实现，和任何消费者无关。

消费位点之间的关系

消费位点初始值指的是消费者分组首次启动消费者消费消息时，服务端保存的消费位点的初始值

消费位点的初始值为消费者首次获取消息时，该时刻队列中的最大消息位点。相当于消费者将从队列中最新的消息开始消费。

若消费者分组的初始消费位点或当前消费位点不符合您的业务预期，您可以通过重置消费位点调整您的消费进度。

初始消费位点不符合需求

消费堆积快速清理

业务回溯，纠正处理

重置到队列中的指定位点

重置到某一时刻对应的消费位点，匹配位点时，服务端会根据自动匹配到该时刻最接近的消费位点

重置消费位点会给系统带来额外处理压力，可能会影响新消息的读写性能。 因此该操作请在适用场景下谨慎执行，并提前做好合理性和必要性评估。

控制消息在重试流程中的状态和变化逻辑

消费失败或者超时后，下一次重新消费的间隔

最大重试次数为3次，则该消息最多可被投递4次，1次为原始消息，3次为重试投递次数。

无序消息

顺序消息

PushConsumer触发消息重试只需要返回消费失败的状态码即可，当出现非预期的异常时，也会被SDK捕获。

SimpleConsumer simpleConsumer = null;
        //消费示例：使用PushConsumer消费普通消息，如果消费失败返回错误，即可触发重试。
        MessageListener messageListener = new MessageListener() {
            @Override
            public ConsumeResult consume(MessageView messageView) {
                System.out.println(messageView);
                //返回消费失败，会自动重试，直至到达最大重试次数。
                return ConsumeResult.FAILURE;
            }
        };

和PushConsumer消费重试策略不同的是，SimpleConsumer消费者的重试间隔是预分配的，每次获取消息消费者会在调用API时设置一个不可见时间参数 InvisibleDuration，即消息的最大处理时长。若消息消费失败触发重试，不需要设置下一次重试的时间间隔，直接复用不可见时间参数的取值。

总结：PushConsumer有一个重试间隔，而SimpleConsumer有一个不可见时间，无需再设一个重试间隔

您预设消息处理耗时最多20 ms，但实际业务中20 ms内消息处理不完，您可以修改消息不可见时间，延长消息处理时间，避免消息触发重试机制

消息不可见时间修改后立即生效，即从调用API时刻开始，重新计算消息不可见时间

默认16次

消息重试间隔=不可见时间－消息实际处理时长

例如，消息不可见时间为30 ms，实际消息处理用了10 ms就返回失败响应，则距下次消息重试还需要20 ms，此时的消息重试间隔即为20 ms

若直到30 ms消息还未处理完成且未返回结果，则消息超时，立即重试，此时重试间隔即为0 ms

SimpleConsumer 触发消息重试只需要等待即可

//消费示例：使用SimpleConsumer消费普通消息，如果希望重试，只需要静默等待超时即可，服务端会自动重试。
        List<MessageView> messageViewList = null;
        try {
            messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                //如果处理失败，希望服务端重试，只需要忽略即可，等待消息再次可见后即可重试获取。
            });
        } catch (ClientException e) {
            //如果遇到系统流控等原因造成拉取失败，需要重新发起获取消息请求。
            e.printStackTrace();
        }

合理重试，避免因限流等诉求触发消费重试

合理控制重试次数，避免无限重试

建议在存储成本可控的前提下，尽可能延长消息存储时长。延长消息存储时长，可以为紧急故障恢复、应急问题排查和消息回溯带来更多的可操作空间。

注意

用于消息过滤，只支持一个消息设置一个tag

这条消息的业务关键词

服务器会根据 keys 创建哈希索引，设置后，可以在 Console 系统根据 Topic、Keys 来查询消息，
由于是哈希索引，请尽可能保证 key 唯一，例如订单号，商品 Id 等。

系统保留字段

默认值0，完全由应用设置

默认值0，0表示不延迟；
大于0会延时特定时间（有个延迟级别对应的延迟时间）才被消费

原理是利用了properties属性机制

默认true，表示消息是否在服务器落盘后才应答

Topic 与 Tag 都是业务上用来归类的标识，区别在于 Topic 是一级分类，而 Tag 可以理解为是二级分类。使用 Tag 可以实现对 Topic 中的消息进行过滤。

消息类型是否一致

业务是否有关联

消息优先级是否一致

消息量级是否相当

每个消息可以在业务层面的设置唯一标识码 keys 字段，方便将来定位消息丢失问题

Broker 端会为每个消息创建索引（哈希索引），应用可以通过 topic、key 来查询这条消息内容，以及消息被谁消费。
由于是哈希索引，请务必保证 key 尽可能唯一，这样可以避免潜在的哈希冲突。

为了支持高并发和水平扩展，需要对 Topic 进行分区，在 RocketMQ 中这被称为队列，
一个 Topic 可能有多个队列，并且可能分布在不同的 Broker 上

一般来说一条消息，如果没有重复发送（比如因为服务端没有响应而进行重试），则只会存在在 Topic 的其中一个队列中，消息在队列中按照先进先出的原则存储，每条消息会有自己的位点，每个队列会统计当前消息的总条数，这个称为最大位点 MaxOffset；队列的起始位置对应的位置叫做起始位点 MinOffset。队列可以提升消息发送和消费的并发度。

消息的发送者，在不同的场景中，需要使用不同类型的消息进行发送

延时消息

顺序消息

批量消息

事务消息

生产环境中不同消息类型需要使用不同的主题，不要在同一个主题内使用多种消息类型，这样可以避免运维过程中的风险和错误

是什么 & 优缺点：收到服务端同步响应之后才发下一条消息，阻塞的

应用场景：重要的通知消息、短消息通知

注意：同步发送方式请务必捕获发送异常，并做业务侧失败兜底逻辑，如果忽略异常则可能会导致消息未成功发送的情况。

是什么 & 优缺点：不等服务端返回响应，接着发送下一条消息，不阻塞

注意：异步发送需要实现异步发送回调接口（SendCallback）。

应用场景：一般用于链路耗时较长，对响应时间较为敏感的业务场景。例如，视频上传后通知启动转码服务，转码完成后通知推送转码结果等

优缺点：只发一次，没有响应结果，不可靠，但一般在毫秒级发送完。

应用场景：适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集

对于一个指定的Topic，消息严格按照先进先出（FIFO）的原则进行消息发布和消费，即先发布的消息先消费，后发布的消息后消费

支持分区顺序消息

同时满足生产顺序性和消费顺序性

RocketMQ 通过生产者和服务端的协议保障 单个生产者 串行 地发送消息，并按序存储和持久化

同一个ShardingKey的消息会被分配到同一个队列中

单一生产者

串行发送

在有序事件处理、撮合交易、数据实时增量同步等场景下，异构系统间需要维持强一致的状态同步，上游的事件变更需要按照顺序传递到下游进行处理

例如创建订单的场景，需要保证同一个订单的生成、付款和发货，这三个操作被顺序执行。如果是普通消息，订单A的消息可能会被轮询发送到不同的队列中，不同队列的消息将无法保持顺序，而顺序消息发送时将ShardingKey相同（同一订单号）的消息序路由到一个逻辑队列中。

2种模式

在一些对数据一致性有强需求的场景，可以用 Apache RocketMQ 事务消息来解决，从而保证上下游数据的一致性

LocalTransactionState.COMMIT_MESSAGE 提交事务

LocalTransactionState.ROLLBACK_MESSAGE 回滚事务

LocalTransactionState.UNKNOW

此外，需要注意的是事务消息的生产组名称 ProducerGroupName不能随意设置。事务消息有回查机制，回查时Broker端如果发现原始生产者已经崩溃，则会联系同一生产者组的其他生产者实例回查本地事务执行情况以Commit或Rollback半事务消息。

集群模式下，同一个消费组内的消费者会分担收到的全量消息，这里的分配策略是怎样的？如果扩容消费者是否一定能提升消费能力？

平均分配策略（默认）

机房优先分配策略

一致性hash分配策略

如果消费者发生崩溃或有新的消费者加入群组，就会触发重平衡

重平衡完成后，每个消费者可能会分配到新的队列，而不是之前处理的队列

为了能继续之前的工作，消费者需要读取每个队列最后一次的提交的消费位点，然后从消费位点处继续拉取消息

但在实际执行过程中，由于客户端提交给服务端的消费位点并不是实时的，所以重平衡就可能会导致消息少量重复

优点：实时性高

缺点：但如果Consumer没有做好流控，Broker推送大量消息给Consumer时，就会导致Consumer消息堆积甚至崩溃

优点：Consumer能根据自己的能力消费

缺点：拉取频率需要自己控制，太频繁会给Broker、Consumer都造成压力；间隔太久又容易造成消费不及时

顺序消费设置与并发消费API层面只有一处不同，在注册消费回调接口时传入MessageListenerOrderly接口的实现

是什么：用于对某个Topic下的消息进行分类，消费者需根据已经指定的Tag来进行订阅

tag表达式

注意：同一个消费者多次订阅某个Topic下的Tag，以最后一次订阅为准

是什么：在消息发送时设置消息的Tag或自定义属性，消费者订阅时使用SQL语法设置过滤表达式，根据自定义属性或Tag过滤消息。

注意：Tag属于一种特殊的消息属性，在SQL语法中，Tag的属性值为TAGS。 开启属性过滤首先要在Broker端设置配置enablePropertyFilter=true，该值默认为false

例子：

若Consumer消费某条消息失败，则RocketMQ会在重试间隔时间后，将消息重新投递给Consumer消费，若达到最大重试次数后消息还没有成功被消费，则消息将被投递至死信队列

消息重试只针对集群消费模式生效；广播消费模式不提供失败重试特性，即消费失败后，失败消息不再重试，继续消费新的消息

顺序消费消费失败后会先在客户端本地重试直到最大重试次数，这样可以避免消费失败的消息被跳过，消费下一条消息而打乱顺序消费的顺序

并发消费消费失败后会将消费失败的消息重新投递回服务端，再等待服务端重新投递回来，在这期间会正常消费队列后面的消息。

是什么：消息达到最大重试次数，若仍然消费失败，则该消息被放入死信队列中。死信队列是死信Topic下分区数唯一的单独队列

死信Topic名称为%DLQ%ConsumerGroupName

死信队列的消息将不会再被消费

拉取消息调用的是轮询poll接口，默认自动提交位点

同一个消费组下的多个LitePullConsumer会负载均衡消费

没有自动的负载均衡机制