消费一批消息的个数是 可配置的 consumeMessageBatchMaxSize = 1， 默认批量个数为一个 ConsumeRequest 任务 run 方法执行 判断 proccessQueue 是否被 droped 的， 废弃直接返回，不在消费消息 构建并行消费上下文 给消息设置消费失败时候的 retry topic，当消息发送失败的时候发送到 Messagequeue 组成。 ConsumeRequest 任务的 run 方法 判断 proccessQueue 是否被 droped 的， 废弃直接返回，不在消费消息 每个 messagequeue 都会生成一个队列锁来保证在当前 consumer 内，同一个队列串行 消费， 判断 processQueue 的 lock 属性是否为 true，lock 属性是否过期，如果为 checkpoint 作用是当异常恢复时需要根据 checkpoint 点来恢复消息 f) 加载索引服务 indexService g) recover 尝试数据恢复 判断是否是正常恢复，系统启动的启动存储服务(DefaultMessageStore)的时候会创 建一个临时文件 abort, 当系统正常关闭的时候会把这个文件删掉 ，这个类似在 linux 下打开 vi

路由消息、消息发送高可用的实现原理，建议查阅笔者的书籍《 RocketMQ 技术内幕》第二、三章。 Step1：在 Broker 启动流程中，会构建 TopicConfigManager 对象，其构造方法中 首先会判断是否开启了允许自动创建主题，如果启用了自动创建主题，则向 topicConfigT able 中添加默认主题的路由信息。 TopicConfigManager 构造方法： 本文来自『中间件兴趣 从拉取呢？ 温馨提示：本节同样不会详细整个流程，只会点出其关键点，如果想详细了解消息拉取、 消息消费等核心流程，建议大家查阅笔者所著的《RocketMQ 技术内幕》。 在 RocketMQ 中判断是从主拉取，还是从从拉取的核心代码如下： DefaultMessageStore#getMessage long diff = maxOffsetPy - maxPhyOffsetPulling; 通过本文的阅读，您将获得如下信息： 1. RocketMQ 消费 TPS 的收集与计算逻辑。 2. RocketMQ 监控指标的设计思路。 3. RocketMQ 主从同步，消费者从主服务器拉取还是从从服务器拉取的判断逻辑。 本文来自『中间件兴趣圈』公众号，仅作技术交流，未授权任何商业行为。 51 > 1.5 踩坑记：rocketmq-console 消费 TPS 为 0，但消息积压数却在降低是个什么“鬼”

要在业务局面去重，有以下几种去重方式 1. 将消息的唯一键，可以是 msgId，也可以是消息内容中的唯一标识字段，例如订单 Id 等，消费乀前判断是否在 Db 戒 Tair(全尿 KV 存储)中存在，如果丌存在则揑入，幵消费，否则跳过。（实际过程要考虑原子性问题，判断 是否存在可以尝试揑入，如果报主键冲突，则揑入失败，直接跳过） msgId 一定是全尿唯一标识符，但是可能会存在同样的消息有两个丌同 1，即一次只消费一条消息，例如设置为 N，那举每次消费的 消息数小亍等亍 N。 14.3.3 跳过非重要消息 収生消息堆积时，如果消费速度一直追丌上収送速度，可以选择丢弃丌重要的消息 如何判断消费収生了堆积？ public ConsumeConcurrentlyStatus consumeMessage(// List msgs

CREATE_COMPLETE，此时堆栈创建成功。Apache RocketMQ 集群以及其他需要的组 件均已创建完毕。下图列出了默认设置部署完成后在 EC2 Console 的一个 截图，可以从 Name 属性判断这个 Instance 安装的是哪种类型的节点。 Page 13 of 18 使⽤ Apache RocketMQ Apache RocketMQ 在 EC2

CREATE_COMPLETE，此时堆栈创建成功。Apache RocketMQ 集群以及 其他需要的组件均已创建完毕。下图列出了默认设置部署完成后在 EC2 Console 的一 个 截图，可以从 Name 属性判断这个 Instance 安装的是哪种类型的节点。 使用 Apache RocketMQ Apache RocketMQ 在 EC2 上部署所在目录说明： Nameserver