理解rocketmq底层代码导览

RocketMQ 的核心代码主要集中在rocketmq-store存储、rocketmq-broker服务端逻辑和rocketmq-client客户端逻辑三个模块。以下是核心链路的底层实现代码入口和方法1. 发送端消息路由与写入Producer 发送消息时主要逻辑在DefaultMQProducerImpl。sendDefaultImpl(...): 这是发送消息的总入口。它负责完成路由选择、消息预处理、重试逻辑。tryToFindTopicPublishInfo(...): 核心逻辑。从 NameServer 获取 Topic 的路由信息包含有哪些 Queue。selectOneMessageQueue(...):顺序消息的核心。底层会调用MessageQueueSelector.select(...)方法。如果传入了MessageQueueSelector它就会根据传入的orderId计算出该消息应去的MessageQueue并将该 Queue 绑定到发送请求中。2. Broker 端消息落盘核心中的核心这是 RocketMQ 性能的基石主要在DefaultMessageStore和CommitLog类中。DefaultMessageStore.putMessage(...): Broker 接收消息的总入口。CommitLog.putMessage(...):最关键的方法。顺序写Append它通过MappedFile内存映射文件将消息直接追加到 CommitLog 的末尾。锁机制为了保证同一队列的写入顺序它会使用putMessageLock默认是ReentrantLock或SpinLock。底层物理顺序消息会被序列化为二进制流按顺序写进mappedFile.appendMessage(...)。ConsumeQueue.putMessagePositionInfo(...):写入 CommitLog 后Broker 会异步地ReputMessageService将消息的索引信息Offset、Size、TagHash更新到对应的ConsumeQueue文件中。ConsumeQueue是“逻辑队列”消费者其实读的是它。3. 消费端顺序与加锁ConsumeOrderly主要在rocketmq-client的ConsumeMessageOrderlyService类中。ConsumeMessageOrderlyService.consumeMessageDirectly(...): 顺序消费的入口。processQueue.lock():分布式锁的实现客户端在消费前会通过RebalanceImpl向 Broker 发送LOCK_BATCH_MQ请求。Broker 端会记录“这个队列当前被这个 ConsumerID 锁住了”。synchronized (processQueue.getLockObject()):本地锁的实现这是防止同一个 Consumer 内部多线程乱跑的关键。processQueue是每个 Queue 在客户端的缓存它持有一个lockObject对象。通过synchronized这个对象保证同一个队列在本地 JVM 中永远是单线程串行消费的。messageListener.consumeMessage(...): 这是开发者实现的接口。如果该方法返回SUSPEND_CURRENT_QUEUE_A_MOMENTRocketMQ 内部会将消息重新放回该队列的ProcessQueue缓存的头部并设置一个等待时间从而实现“阻塞式重试”。4. 关键数据结构类值得阅读源码的类如果想去看源码建议从这些类开始CommitLog.java:查看putMessage方法会明白为什么 RocketMQ 的写入速度这么快顺序 I/O MappedByteBuffer。ConsumeQueue.java:这是 RocketMQ 的索引文件了解它就理解了为什么能快速按 Offset 查找消息。RebalanceImpl.java:这里是负载均衡的核心。它决定了哪个 Queue 分配给哪个 Consumer。它是“顺序消费”的前提。ProcessQueue.java:这是 Consumer 端的“缓冲队列”它是顺序消费时处理锁、处理消费进度、处理积压的核心结构。DefaultMessageStore.java:这是 Broker 的心脏它串联了存储、索引、重平衡等所有流程。总结底层实现的核心思想存储层利用磁盘顺序写和**内存映射mmap**来保证吞吐量。逻辑层顺序通过Broker 端的写入锁和Consumer 端的ProcessQueue对象锁人为地在分布式系统中构筑了一个“排队通道”。