面试题：消息队列之RocketMQ在复杂电商架构下的专家级应用问题

Question

假设电商系统采用微服务架构，涉及多个子系统之间通过RocketMQ进行消息交互，在面对部分子系统不可用时，如何设计RocketMQ的消息补偿机制和流量控制策略，以保障整个电商业务的最终一致性和稳定性？

Answer 1

消息补偿机制设计

1. 消息持久化
   • RocketMQ本身支持消息持久化到磁盘，确保即使Broker重启，消息也不会丢失。通过合理配置存储路径、刷盘策略（同步刷盘或异步刷盘），保证消息的可靠存储。例如，对于关键的订单创建、支付成功等消息，采用同步刷盘策略，以确保消息写入磁盘成功后才返回确认。
2. 消息重试机制
   • 生产者重试：当生产者发送消息失败时（比如网络异常、Broker不可用等），开启重试机制。RocketMQ的生产者客户端支持设置重试次数，一般可根据业务场景设置3 - 5次重试。例如，在发送订单创建消息时，如果第一次发送失败，生产者可等待一定时间（如100ms、200ms等，可根据业务情况调整）后进行重试。
   • 消费者重试：消费者消费消息失败时，RocketMQ提供自动重试机制。默认情况下，消息会重试16次，每次重试间隔时间逐渐变长（如10s、30s、1min等）。对于业务异常导致的消费失败，消费者可捕获异常并进行相应处理，如记录日志，然后抛出异常让RocketMQ进行重试。例如，在处理订单支付消息时，如果由于库存扣减失败导致消费失败，可记录库存相关信息及失败原因，让RocketMQ重试。
3. 死信队列
   • 对于多次重试后仍然失败的消息，将其发送到死信队列。在死信队列中，可通过监控工具定期查看死信消息，分析失败原因。例如，若发现大量订单支付成功但库存扣减失败的死信消息，可能是库存服务出现性能瓶颈或逻辑错误，开发人员可针对性地进行排查和修复。同时，也可以编写定时任务从死信队列中重新消费这些消息，尝试再次处理。
4. 消息幂等性处理
   • 业务层面幂等：在业务代码中设计幂等性逻辑。例如，在处理订单创建消息时，通过订单号作为唯一标识，每次接收到订单创建消息，先查询数据库中是否已存在该订单号的订单记录，如果存在则不再重复创建订单，直接返回成功。
   • MQ层面幂等：RocketMQ支持生产者发送消息的幂等性，通过设置Producer的enableMsgTrace和customizedTraceTopic属性，开启消息轨迹功能，结合消息的唯一标识（如msgId），在Broker端和客户端实现幂等性校验，避免重复消费。

流量控制策略设计

1. 生产者流量控制
   • 令牌桶算法：在生产者端应用令牌桶算法。例如，设定一个令牌桶容量为100，每秒生成10个令牌。生产者每次发送消息时，需要从令牌桶中获取一个令牌，如果令牌桶中没有令牌，则等待令牌生成或者丢弃消息（根据业务策略决定）。这样可以控制生产者发送消息的频率，避免瞬间大量消息涌入Broker，导致Broker压力过大。
   • 动态调整发送速率：根据Broker的负载情况动态调整生产者的发送速率。可以通过监控Broker的CPU使用率、内存使用率、网络带宽等指标，当发现Broker负载过高时，生产者降低发送消息的频率；当Broker负载降低时，适当提高发送频率。例如，使用Prometheus和Grafana搭建监控系统，实时监控Broker指标，并通过接口通知生产者调整发送速率。
2. 消费者流量控制
   • 限流队列：在消费者端设置限流队列，例如使用LinkedBlockingQueue，设定队列容量为1000。当队列满时，新的消息不再进入队列，直接丢弃（可记录丢弃日志），避免消费者因处理能力不足导致内存溢出。消费者从队列中取出消息进行处理，根据处理能力动态调整队列的消费速度。
   • 动态线程池：根据消息队列的长度动态调整消费者线程池的大小。当队列长度超过一定阈值（如800）时，增加线程池中的线程数量，提高消费速度；当队列长度低于一定阈值（如200）时，减少线程池中的线程数量，节省系统资源。例如，使用ThreadPoolExecutor实现动态线程池，通过监控队列长度来调用setCorePoolSize和setMaximumPoolSize方法调整线程池大小。
3. Broker流量控制
   • Broker配置限流：在RocketMQ的Broker配置文件中设置相关限流参数。例如，通过broker.conf文件中的maxSendMessageSize参数限制单个消息的最大大小，避免过大的消息占用过多网络带宽和Broker存储资源。同时，可以设置defaultTopicQueueNums参数，合理分配Topic的队列数量，根据业务流量情况进行调整，以平衡负载。
   • 基于Topic的限流：对不同的Topic设置不同的限流策略。例如，对于订单创建、支付等关键业务的Topic，可设置较低的限流阈值，优先保障其消息处理的稳定性；对于一些非关键业务（如用户评论、积分变更等）的Topic，设置相对较高的限流阈值，以充分利用系统资源。可通过管理控制台或API对不同Topic的限流策略进行动态调整。