[大数据] RocketMQ学习笔记（3）

一、使用RocketMQ如何保证消息不丢失？

我们考虑一个通用的 MQ 场景：其中， 1 ， 2 ， 4 三个场景都是跨网络的，而跨网络就肯定会有丢消息的可能。然后关于3 这个环节，通常 MQ 存盘时都会先写入操作系统的缓存 page cache 中，然后再由操作系统异步的将消息写入硬盘。这个中间有个时间差，就可能会造成消息丢失。如果服务挂了，缓存中还没有来得及写入硬盘的消息就会丢失。这个是MQ 场景都会面对的通用的丢消息问题。那我们看看用 Rocket 时要如何解决这个问题。

1、那些环境会有可能丢消息？

这个结论比较容易理解，因为 RocketMQ 的事务消息机制就是为了保证零丢失来设计的，并且经过阿里的验证，肯定是非常靠谱的。

但是如果深入一点的话，我们还是要理解下这个事务消息到底是不是靠谱。我们以最常见的电商订单场景为例，来简单分析下事务消息机制如何保证消息不丢失。我们看下下面这个流程图：

事务消息机制的作用

整体来说，在订单这个场景下，消息不丢失的问题实际上就还是转化成了下单这个业务与下游服务的业务的分布式事务一致性问题。而事务一致性问题一直以来都是一个非常复杂的问题。而RocketMQ 的事务消息机制，实际上只保证了整个事务消息的一半，他保证的是订单系统下单和发消息这两个事件的事务一致性，而对下游服务的事务并没有保证。但是即便如此，也是分布式事务的一个很好的降级方案。

2、RocketMQ消息零丢失方案？

完整分析过后，整个 RocketMQ 消息零丢失的方案其实挺简单

生产者使用事务消息机制。

Broker 配置同步刷盘 +Dledger 主从架构

消费者不要使用异步消费。

整个 MQ 挂了之后准备降级方案

那这套方案是不是就很完美呢？其实很明显，这整套的消息零丢失方案，在各个环节都大量的降低了系统的处理性能以及吞吐量。在很多场景下，这套方案带来的性能损失的代价可能远远大于部分消息丢失的代价。所以，我们在设计RocketMQ 使用方案时，要根据实际的业务情况来考虑。例如，如果针对所有服务器都在同一个机房的场景，完全可以把Broker 配置成异步刷盘来提升吞吐量。而在有些对消息可靠性要求没有那么高的场景，在生产者端就可以采用其他一些更简单的方案来提升吞吐，而采用定时对账、补偿的机制来提高消息的可靠性。而如果消费者不需要进行消息存盘，那使用异步消费的机制带来的性能提升也是非常显著的。

总之，这套消息零丢失方案的总结是为了在设计 RocketMQ 使用方案时的一个很好的参考。

二、使用RocketMQ如何保证消息有序？

1、为什么要保证消息有序？

例如如果我们有个大数据系统，需要对业务系统的日志进行收集分析，这时候为了减少对业务系统的影响，通常都会通过MQ 来做消息中转。而这时候，对消息的顺序就有一定的要求了。例如我们考虑下面这一系列的操作。

1. 用户的积分默认是 0 分，而新注册用户设置为默认的 10 分。

2. 用户有奖励行为，积分 +2 分。

3. 用户有不正当行为，积分 -3 分。

这样一组操作，正常用户积分要变成 9 分。但是如果顺序乱了，这个结果就全部对不了。这时，就需要对这一组操作，保证消息都是有序的。

2、如何保证消息有序？

MQ 的顺序问题分为全局有序和局部有序。

全局有序：整个 MQ 系统的所有消息严格按照队列先入先出顺序进行消费。

局部有序：只保证一部分关键消息的消费顺序。

首先我们需要分析下这个问题，在通常的业务场景中，全局有序和局部有序哪个更重要？其实在大部分的MQ 业务场景，我们只需要能够保证局部有序就可以了。例如我们用 QQ 聊天，只需要保证一个聊天窗口里的消息有序就可以了。而对于电商订单场景，也只要保证一个订单的所有消息是有序的就可以了。至于全局消息的顺序，并不会太关心。而通常意义下，全局有序都可以压缩成局部有序的问题。例如以前我们常用的聊天室，就是个典型的需要保证消息全局有序的场景。但是这种场景，通常可以压缩成只有一个聊天窗口的QQ 来理解。即整个系统只有一个聊天通道，这样就可以用 QQ 那种保证一个聊天窗口消息有序的方式来保证整个系统的全局消息有序。然后落地到 RocketMQ 。通常情况下，发送者发送消息时，会通过 MessageQueue 轮询的方式保证消息尽量均匀的分布到所有的MessageQueue 上，而消费者也就同样需要从多个 MessageQueue 上消费消息。而MessageQueue 是 RocketMQ 存储消息的最小单元，他们之间的消息都是互相隔离的，在这种情况下，是无法保证消息全局有序的。而对于局部有序的要求，只需要将有序的一组消息都存入同一个MessageQueue 里，这样MessageQueue的 FIFO 设计天生就可以保证这一组消息的有序。 RocketMQ 中，可以在发送者发送消息时指定一个MessageSelector 对象，让这个对象来决定消息发入哪一个 MessageQueue 。这样就可以保证一组有序的消息能够发到同一个MessageQueue 里。另外，通常所谓的保证Topic 全局消息有序的方式，就是将 Topic 配置成只有一个 MessageQueue 队列 (默认是 4 个 ) 。这样天生就能保证消息全局有序了。这个说法其实就是我们将聊天室场景压缩成只有一个聊天窗口的QQ 一样的理解方式。而这种方式对整个 Topic 的消息吞吐影响是非常大的，如果这样用，基本上就没有用MQ 的必要了。

三、使用RocketMQ如何处理积压消息？

1、如何确定RocketMQ有大量的消息积压？

在正常情况下，使用 MQ 都会要尽量保证他的消息生产速度和消费速度整体上是平衡的，但是如果部分消费者系统出现故障，就会造成大量的消息积累。这类问题通常在实际工作中会出现得比较隐蔽。例如某一天一个数据库突然挂了，大家大概率就会集中处理数据库的问题。等好不容易把数据库恢复过来了，这时基于这个数据库服务的消费者程序就会积累大量的消息。或者网络波动等情况，也会导致消息大量的积累。这在一些大型的互联网项目中，消息积压的速度是相当恐怖的。所以消息积压是个需要时时关注的问题。对于消息积压，如果是RocketMQ 或者 kafka 还好，他们的消息积压不会对性能造成很大的影响。而如果是RabbitMQ 的话，那就惨了，大量的消息积压可以瞬间造成性能直线下滑。对于RocketMQ 来说，有个最简单的方式来确定消息是否有积压。那就是使用 web 控制台，就能直接看到消息的积压情况。在Web 控制台的主题页面，可以通过 Consumer 管理按钮实时看到消息的积压情况。

?

2、如何处理大量积压的消息？

其实我们回顾下 RocketMQ 的负载均衡的内容就不难想到解决方案。

如果 Topic 下的 MessageQueue 配置得是足够多的，那每个 Consumer 实际上会分配多个 MessageQueue来进行消费。这个时候，就可以简单的通过增加 Consumer 的服务节点数量来加快消息的消费，等积压消息消费完了，再恢复成正常情况。最极限的情况是把Consumer 的节点个数设置成跟MessageQueue的个数相同。但是如果此时再继续增加 Consumer 的服务节点就没有用了。而如果Topic 下的 MessageQueue 配置得不够多的话，那就不能用上面这种增加 Consumer 节点个数的方法了。这时怎么办呢？这时如果要快速处理积压的消息，可以创建一个新的Topic ，配置足够多的MessageQueue。然后把所有消费者节点的目标 Topic 转向新的 Topic ，并紧急上线一组新的消费者，只负责消费旧Topic 中的消息，并转储到新的 Topic 中，这个速度是可以很快的。然后在新的 Topic 上，就可以通过增加消费者个数来提高消费速度了。之后再根据情况恢复成正常情况。

四、RocketMQ的消息轨迹

1、RocketMQ消息轨迹数据的关键属性

2、消息轨迹配置？

3、消息轨迹数据存储？

默认情况下，消息轨迹数据是存于一个系统级别的 Topic ,RMQ_SYS_TRACE_TOPIC 。这个 Topic 在Broker节点启动时，会自动创建出来。

另外，也支持客户端自定义轨迹数据存储的 Topic 。

traceTopicEnable=true

在客户端的两个核心对象 DefaultMQProducer 和 DefaultMQPushConsumer ，他们的构造函数中，都有两个可选的参数来打开消息轨迹存储。

enableMsgTrace ：是否打开消息轨迹。默认是 false 。

customizedTraceTopic ：配置将消息轨迹数据存储到用户指定的 Topic 。