rocketmq3 java - 锐客网

【rocketmq3】从节点的分流判断

针对消息堆积量过大会切换到Slave进行查询。maxOffsetPy 为当前最大物理偏移量，maxPhyOffsetPulling 为本次消息拉取最大物理偏移量，他们的差即可表示消息堆积量TOTAL_PHYSICAL_MEMORY_SIZE 表示当前系统物理内存，accessMessageInMemoryMaxRatio 的默认值为 40，以上逻辑即可算出当前消息堆积量是否大于物理内存的 40%，如果大于则将 suggestPullingFromSlave 设置为 true。 //https://www.jianshu.com/p/cd138e67dca0

master宕机

1情况：nameServer未检测到broker掉线，无论客户端是否发生了rebalance，producer、consumer不知道broker挂了 2情况：nameServer检测到了broker掉线，但是客户端未发生rebalance，producer、consumer不知道broker挂了。 3情况：nameServer检测到了broker掉线，客户端发生了rebalance，producer、consumer感知到了broker掉线。# nameServer管理broker。在broker注册的时候，会通过registerBrokerAll把topic信息（包含queue的信息）、地址、brokerId、brokerName上报到nameServer。broker在启动的时候，定时任务调用registerBrokerAll。在更新broker配置、创建topic的时候，也会调用registerBrokerAll。 nameServer启动时会开启一个定时任务，定时检测broker的心跳。如果某broker心跳不在了，会进行移除操作。只有同一个brokerName的所有broker都没了，才会移除对应的queue信息。发送者：在1、2时，对此broker上的queue发送消息时，会失败，同步发送的话，会重试，重试也可能失败，最终会抛出异常。等到在3时，producer端会发现此queue的masterId的broker挂了，就会抛出异常。消费者：在1、2时，拉取此broker上的queue消息时，会失败，在push模式下会稍后接着拉去，等到在3时，consumer会发现此queue的masterId的broker挂了，就会寻找同brokerName的broker的地址拉去消息，也就是会去从拉取消息。对于producer来说，主挂了，再对此broker上的queue发消息都会失败抛异常给用户，用户可以指定queue选择器（或自己实现）来排除对此broker上的queue发送消息。对于consumer来说，主挂了，等到rebanlance后，就会从此broker的从拉取消息。

多线程写磁盘速度是否提升

https://blog.csdn.net/u013043103/article/details/84326462（1）读写最好还是不要多线程，硬盘读写的速度有限，单线程时已经满负荷了，多线程又会增加线程之间的切换，会增加时间。如果想增加读写速度，应该增加硬盘，做raid（2）首先是硬盘的写入是串行的，CPU的计算才是并行的，如果你偏重计算那么多线程能提高，要不怎么叫做并行计算呢；如果侧重存储，除非数据量达到足以体现优势的程度，否则加上线程之间切换的损耗当然会效率更加地下。（3）这个是按照算法来说的，目前来说大多数的算法都是很快的，瓶颈都在磁盘的IO上，我们针对大多数的算法都进行过测试，基本一半以上的时间都耗费在磁盘的IO上。比如我处理一个影像，处理数据用了1分钟，写入图像用了2分钟，那你把你的算法优化的很牛逼，10秒中搞定，你的效率提高了多少，但是如果我多线程写入的话，我效率提高一倍，也就是写入图像用了1分钟，那这个效率明显比你优化你的算法来的实惠。这个东西还是要针对算法来说的。（4）磁盘IO单线程顺序写时最快的，如果多线程写，磁盘的磁头要不断重新寻址，所以写入速度反而会慢。

根据msgid做幂等不可靠

当存在网络波动，网路延时等诸多问题时，消息从客户端发送至服务端过程中，服务端正常写入了commit-log，可在响应客户端（ACK）的时候失败了。导致两条一样的消息内容，却有了不一样的MsgId跟OffsetMsgId，最终它还是重复消费了消费者如果消费失败，调用sendMessageBack方法将消息发给broker，重新消费时的消息msgId不变，offsetMsgId会变(因为新消息储存的地址已变)，uniq_key属性保存原消息msgId

开启临时内存transientStorePoolEnable

mappedFile有两种形式，一种是只有映射内存，写和读都是从映射的内存中读。但是读写都是直接操作pagecache，会导致在大量读冷数据的时候，pagecache吃紧的时候，读和写会相互影响，可能会导致写超时之类（默认超过200ms）。一种是临时内存+映射内存。临时内存是有jvm向操作系统申请的直接内存，直接会锁住，防止内核将此块内存置换到硬盘。这种形式下，写入直接写入临时内存就返回，读的话依旧读映射内存，直接内存会有一个服务commit到映射内存，写入直接内存但是没有commit的数据是不可以读的。 commit的策略，每次在put消息结束后都会唤醒刷盘服务，如果未commit的数据超过leastPages（默认4k）或者距离上次commit的时间超过commitDataThoroughInterval（默认200ms）就会commit一次。开启临时内存会优化读写的冲突，提升性能，等于批量写pagecache，尤其在读冷数据pachecache吃紧的情况下。但是写入消息不频繁（200ms内，消息量达不到4k），消息会产生较为明显额外的延迟。在写入压力正常的情况下，基本不会有额外的延迟。

默认路由

默认路由TBW102说明一下broker如果开启了autoCreateTopicEnable(默认就是true)，broker会向namesrv注册默认路由topic：TBW102当producer从namesrv取topic路由如果不存在，就会取TBW102的路由信息，查看哪些broker支持自动创建topic，然后构造新的topic路由信息。这里有一个坑点，如果使用默认路由，可能出现topic只在一个broker中创建，例如如下情况一开始broker-a和broker-b支持自动创建路由，producer本地更新了本地路由表[topicnew->broker-a; topicnew->broker-b]，然后发送消息给broker-a，broker-a更新了topicnew到namesrv，这时的namesrv上只有[topicnew->broker-a]一个路由信息，如果一段时间producer没有发送消息给broker-b，然后从namesrv更新了topicnew的路由信息，则最终就会出现新增的topic只被路由到一个broker中，没有高可用。所以在消息不是很频繁的情况下，最好手动创建topic。

起始位点逻辑

默认情况下为：最新位点所属的消费者组是新上线的，订阅的消息，最早的消息(offset=0）还在内存中。rocketmq的设计者认为，你这是一个新上线的业务，会强制从第一条消息开始消费。如果订阅的消息，已经产生了过期消息。那么才会从我们这个client启动的时间点开始消费。设置为起始位点：客户端计算逻辑会为0，然后服务端发现位点比最小位点要小，就会更正位点，设置为最小位点设置时间戳：如果消费者组以前消费过某个topic，setConsumeFromWhere这个参数是不起效的。只要broker找的到消费位点，就是按照broker的来RebalancePushImpl#computePullFromWhereWithException ConsumerManageProcessor#queryConsumerOffset DefaultMessageStore#getMessage

事务回查实现

TransactionalMessageCheckService类中，此类包含一个线程，此线程默认每分钟触发一次事务检查，在其onWaitEnd方法中，实际上还是调用了TransactionalMessageService的check方法。大致的回查逻辑：Half Topic 对应队列中存放着 prepare 消息，Operation Topic 对应的队列则存放了 prepare message 对应的 commit/rollback 消息，消息体中则是 prepare message 对应的 offset，服务端通过比对两个队列的差值来找到尚未提交的超时事务，进行回查。https://www.jianshu.com/p/feda710d9716

刷盘服务

https://blog.csdn.net/yecong111/article/details/103858172

位点保存逻辑

每个queue都会有ProcessQueue来维护消费情况每次拉取一批消息，都会把消息放到一个treeMap中，key为offset，value为msg 每次消费完某条消息，就会把这条信息在treeMap中移除。然后将内存中的offsetTable的信息更新，更新为treeMap的树顶。然后在MQClientInstance启动的时候会启动一个定时任务，定时的上报位点到远程。根据queue获取brokerName，然后再获取地址，然后发送请求。在pull消息的时候，如果本地位点大于0，就设置可以提交位点，把本地位点提交上去。broker如果是从，不会理会，如果是主，就会更新位点到内存。broker端内存维护各个消费组的消费位点，定时任务持久化（BrokerController#init开启的定时任务）。ps：位点的这条消息是没有消费的，需要消费。

延迟消息

延迟队列的核心思路是： Producer发送消息是，指定delayLevel；或者Consumer消费消息时，返回RECONSUME_LATER，或者主动的sendMessageBack(…，int delayLevel)时，会将消息发回给Broker，Broker对消息做个封装，指定topic为SCHEDULE_TOPIC_XXXX，QueudId=delayLevel-1，若未指定delayLevel，默认是ReConsumeTimes + 3，将封装后的消息存入CommitLog，ReputMessageService为其生成PositionInfo，tagsCode存储延时投递时间，存入"SCHEDULE_TOPIC_XXXX"的ConsumeQueue中。delayLevel有16个，因此最多情况下SCHEDULE_TOPIC_XXXX会有16个ConsumeQueue。Broker启动时，ScheduleMessageService会启动16个线程对应16个delayLevel的读取服务，有序的读取ConsumeQueue里的PositionInfo。ScheduleMessageService会在 [当前时间<=延时投递时间] 时从CommitLog中提取这消息，去除封装，抹去delayLevel属性，重新存入CommitLog，并马上更新延时投递偏移量dealyOffset。ReputMessageService再次为当前消息生成PositionInfo，因为不存在delayLevel，PositionInfo存入Topic为%RETRY%+consumeGroup，queueId为0的ConsumeQueue中。每个消费者在启动时都订阅了自身消费者组的重试队列，当重试队列里有位置信息时，拉取相应消息进行重新消费。消息的第一次重试会发回给原始的消费者(执行sendMessageBack的消费者)，之后的多次重试统一由订阅了QueueId = 0 的消费者消费。消费者发回消息时，可以指定延迟级别，默认级别：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，也就是说delayLevel = 3代表延迟10秒后重投递，最大重试次数16对应着2h后投递，每多消费一次投递时间就增长到下个阶段。当延迟级别delayLevel < 0时，放入Dead Letter Queue。

重置位点

重置位点的执行顺序按照admin 到 broker 到 consumer的顺序依次触发，admin负责构建参数通知broker，broker负责查询consumeQueue的具体位移，broker负责通知consumer进行位移重置。根据时间戳查找consumeQueue对应的位移，然后由broker通知consumer来持久化消费位移，最终会持久化到broker的消费位移。重置位点操作本质上是在consumer端执行，consumer端负责持久化新的消费位移然后由定时任务通知broker更新消费位移。consumer在整个位移重置过程中会设置ProcessQueue的状态为Dropped，从而阻断消息拉取任务ConsumeRequest的执行阻断消息拉取，其次会在consumer侧修改消费位移通过心跳通知broker修改consumer的消费位移，最后通过重新的rebalance过程开始重新消费消息

主从数据不一致

造成的原因是落盘落后于主从复制，导致当主挂了，可能数据没有落盘，但是这些数据可以从从消费到，而主恢复后，这一部分数据丢失，导致数据不一致。

主从同步

RocketMQ 的主从同步机制如下： (1) slave启动，跟master建立连接 (2) slave 以5秒的间隔，向master拉取消息，如果是第一次拉取的话，先获取本地commitlog文件中最大的偏移量，以该偏移量向服务端拉取消息； (3) master将数据返回给slave (4) slave将数据写入自身commitLog中，更新偏移量；重复以上步骤从服务器定时调度syncAll方法，定时向主服务器同步消费组订阅信息、消费位点、延迟消息位点、topic配置。 BrokerController#handleSlaveSynchronize