网站建设实施方案官网建站多少钱
消息过滤与重试
What is going to happen when consumer listener returns “RECONSUME_LATER”?
在Consumer的消息消费监听器中，如果抛出来RuntimeException并且你自己没有捕获，那也等同于直接返回了“RECONSUME_LATER”
消费失败时，重试消息的处理逻辑：
注意，最原始的消息为1号消息，是存在程序员自定义的业务topic中的，而一旦1号消息消费失败，会被客户端重新发回broker端
broker端接收到这条发回的消息后，先把1号消息的原业务topic更改为 %RETRY%consumer group，然后将此消息的topic改为SCHEDULE_TOPIC_XXXX，将此消息的属性key为PROPERTY_REAL_TOPIC对应的value中保存 %RETRY%consumer group这个topic，并将这条消息写入commitlog成为2号消息（注意，这里还会把消息的retryTimes + 1）
2号消息存储的topic为SCHEDULE_TOPIC_XXXX， 一旦2号消息到达了指定的延迟时间后，会被再次取出成为3号消息，此时将2号消息的属性key为PROPERTY_REAL_TOPIC对应的value中保存的topic值 %RETRY%consumer group取出来，设置为3号消息的topic，并将3号消息重新投递到topic为%RETRY%consumer group对应的队列中去
后续，就可以按照正常的逻辑进行消费了
这里是分为了18个等级，分别对应18个queue，这也就有点类似于kafka中的时间轮，将相同延迟级别的消息放入同一个queue，方便统一管理控制
在kafka中是使用了时间轮，进行了更为精确的时间控制
具体的延迟实现逻辑：
18个延迟级别，分别对应18个ConsumeQueue，并且这18个ConsumeQueue同属于同一个topic：SCHEDULE_TOPIC_XXXX
针对这18个ConsumeQueue，每个都创建了一个专属的延时TimeTask并丢入了一个统一的Timer定时任务实例中，这18个任务初始默认都是1s后执行
每个专属的延时TimeTask的执行逻辑是，
- 先从delayLevel.json中先加载已有的消费进度，从而得到下一次要消费的offset，通过这个offset去对应的ConsumeQueue中拿到对应的索引条目，从中拿到phyOffset、size、tagHash，
- 需要注意的是，如果往Commitlog中写入的延时消息时，ReputMessageService会把写入的ConsumeQueue对应的索引条目中tagHash，改写为该延时消息下次要执行的时间对应的时间戳nextExecTimestamp
- 专属的延时TimeTask拿到该延时消息对应的索引条目的nextExecTimestamp后，与当前时间戳now取一个差值countdown
- 如果countdown<=0，说明当前消息已经到了需要被消费的时候了，那么就把这个索引条目对应的延时消息从commitlog中取出，并将该消息对应的topic从SCHEDULE_TOPIC_XXXX，改成该延时消息的属性key为PROPERTY_REAL_TOPIC对应的value中保存的topic值 %RETRY%consumer group，然后把改完后的3号消息，重新调用DefaultMessageStore#putMessage()方法，把该三3号消息投入Commitlog，后续就进去了正常的消费逻辑
- 如果countdown<=0，说明当前消息需要再等countdown毫秒才可以被消费，此时就重新new出一个延时TimeTask并带上offset，然后把这个TimeTask丢入Timer中，指定再等countdown毫秒执行这个TimeTask
注意，这里有一个兜底策略，就是如果每个延迟队列很长时间都没有新消息进来，那么每个延迟队列对应的TimeTask，也会每隔100ms被丢入Timer中一次。具体逻辑就是，该延迟队列的上一个TimeTask执行过程中发现该延迟队列没有新的延迟消息，则会在最后，往timer中丢入一个TimeTask，并指定这个TimeTask在100ms后执行，以此循环往复
消费过滤
Where does RocketMQ filter messages? Broker or Consumer?
