1. 讲述rocketmq线程模型的设计优势，做到线程池隔离
图解Kafka线程模型及其设计缺陷
2. Broker端busy的原因总结
生产环境中在消息发送过程中偶尔会出现如下4个错误信息之一：
1）[REJECTREQUEST]system busy, start flow control for a while
2）too many requests and system thread pool busy, RejectedExecutionException
3）[PC_SYNCHRONIZED]broker busy, start flow control for a while
4）[PCBUSY_CLEAN_QUEUE]broker busy, start flow control for a while, period in queue: %sms, size of queue: %d
RocketMQ 消息发送system busy、broker busy原因分析与解决方案_中间件兴趣圈-CSDN博客
RocketMQ 消息发送system busy、broker busy原因分析与解决方案_中间件兴趣圈-CSDN博客
3. 发送超时和system busy，broker busy的解决方案
发送超时时：减少消息发送超时时间，增大重试次数，并增加快速失败的最长等待时间
system busy，broker busy主要有三大类原因：
a）PageCache压力大
b）发送线程池挤压的拒绝策略
c）broker端的快速失败
消息发送常见错误与解决方案_中间件兴趣圈-CSDN博客
3. rocketmq实现发送消息的高可用的策略
1. 客户端发送重试
2. broker端配合快速失败策略
Rocketmq TIMEOUT_CLEAN_QUEUE源码追踪_八荒六合唯我独尊-CSDN博客
keys：
a) 从 Broker 端快速失败机制引入的初衷来看，快速失败后会发起重试，除非同一深刻集群内所有的 Broker 都繁忙，不然消息会发送成功，用户是不会感知这个错误的
b)broker端快速失败的原理图
3. RocketMQ 消息发送高可用设计一个非常关键的点，重试机制，其实现是在 for 循环中 使用 try catch 将 sendKernelImpl 方法包裹，就可以保证该方法抛出异常后能继续重试。从上文可知，如果 SYSTEM_BUSY 会抛出 MQBrokerException，但发现只有上述几个错误码才会重试，因为如果不是上述错误码，会继续向外抛出异常，此时 for 循环会被中断，即不会重试。
这里非常令人意外的是连 SYSTEM_ERROR 都会重试，却没有包含 SYSTEM_BUSY，显然违背了快速失败的设计初衷，故笔者断定，这是 RocketMQ 的一个BUG，将 SYSTEM_BUSY 遗漏了，后面与 RocketMQ 核心成员进行过沟通，也印证了这点，后续会提一个 PR，在上面增加一行代码，将 SYSTEM_BUSY 加上即可。
4. TIMEOUT_CLEAN_QUEUE 的解决方法，大家不约而同提出的解决方案是增加 waitTimeMillsInSendQueue 的值，该值默认为 200ms，例如将其设置为 1000s 等等，以前我是反对的，因为我的认知里 Broker 会重试，但现在发现 Broker 不会重试，所以我现在认为该 BUG未解决的情况下适当提高该值能有效的缓解；
5. broker处理队列中的消息的两大类情况：
5.1 broker not busy：
处理消息发送的线程池SendMessageExecutor会从队列SendThreadPoolQueue中获取任务并执行消息写入请求
5.5 broker busy：
a) OSPageCacheBusy时，将SendThreadPoolQueue中的消息写入请求全部弹出，并逐一返回[PCBUSY_CLEAN_QUEUE]broker busy
b) OSPageCache not Busy，但消息写入请求在SendThreadPoolQueue存在时间，超过waitTimeMillsInSendQueue （默认200ms），则将队列中所有超过waitTimeMillsInSendQueue的请求全部弹出，并逐一返回[PCBUSY_CLEAN_QUEUE]broker busy
具体代码见：BrokerFastFailure类
public void start() {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
if (brokerController.getBrokerConfig().isBrokerFastFailureEnable()) {
cleanExpiredRequest();
}
}
}, 1000, 10, TimeUnit.MILLISECONDS);
}
private void cleanExpiredRequest() {
while (this.brokerController.getMessageStore().isOSPageCacheBusy()) {
try {
if (!this.brokerController.getSendThreadPoolQueue().isEmpty()) {
final Runnable runnable = this.brokerController.getSendThreadPoolQueue().poll(0, TimeUnit.SECONDS);
if (null == runnable) {
break;
}
final RequestTask rt = castRunnable(runnable);
rt.returnResponse(RemotingSysResponseCode.SYSTEM_BUSY, String.format("[PCBUSY_CLEAN_QUEUE]broker busy, start flow control for a while, period in queue: %sms, size of queue: %d", System.currentTimeMillis() - rt.getCreateTimestamp(), this.brokerController.getSendThreadPoolQueue().size()));
} else {
break;
}
} catch (Throwable ignored) {
}
}
cleanExpiredRequestInQueue(this.brokerController.getSendThreadPoolQueue(),
this.brokerController.getBrokerConfig().getWaitTimeMillsInSendQueue());
cleanExpiredRequestInQueue(this.brokerController.getPullThreadPoolQueue(),
this.brokerController.getBrokerConfig().getWaitTimeMillsInPullQueue());
cleanExpiredRequestInQueue(this.brokerController.getHeartbeatThreadPoolQueue(),
this.brokerController.getBrokerConfig().getWaitTimeMillsInHeartbeatQueue());
cleanExpiredRequestInQueue(this.brokerController.getEndTransactionThreadPoolQueue(), this
.brokerController.getBrokerConfig().getWaitTimeMillsInTransactionQueue());
}
void cleanExpiredRequestInQueue(final BlockingQueue blockingQueue, final long maxWaitTimeMillsInQueue) {
while (true) {
try {
if (!blockingQueue.isEmpty()) {
final Runnable runnable = blockingQueue.peek();
if (null == runnable) {
break;
}
final RequestTask rt = castRunnable(runnable);
if (rt == null || rt.isStopRun()) {
break;
}
final long behind = System.currentTimeMillis() - rt.getCreateTimestamp();
if (behind >= maxWaitTimeMillsInQueue) {
if (blockingQueue.remove(runnable)) {
rt.setStopRun(true);
rt.returnResponse(RemotingSysResponseCode.SYSTEM_BUSY, String.format("[TIMEOUT_CLEAN_QUEUE]broker busy, start flow control for a while, period in queue: %sms, size of queue: %d", behind, blockingQueue.size()));
}
} else {
break;
}
} else {
break;
}
} catch (Throwable ignored) {
}
}
}
4. 为什么明明集群中有多台Broker服务器，autoCreateTopicEnable设置为true，表示开启Topic自动创建，但新创建的Topic的路由信息只包含在其中一台Broker服务器上，这是为什么呢？
期望值：为了消息发送的高可用，希望新创建的Topic在集群中的每台Broker上创建对应的队列，避免Broker的单节点故障。
RocketMQ实战：生产环境中，autoCreateTopicEnable为什么不能设置为true
这里有三个关键点：
-
启用autoCreateTopicEnable创建主题时，在Broker端创建主题的时机为，消息生产者往Broker端发送消息时才会创建。
-
然后Broker端会在一个心跳包周期内，将新创建的路由信息发送到NameServer，于此同时，Broker端还会有一个定时任务，定时将内存中的路由信息，持久化到Broker端的磁盘上。
-
消息发送者会每隔30s向NameServer更新路由信息，如果消息发送端一段时间内未发送消息，就不会有消息发送集群内的第二台Broker，那么NameServer中新创建的Topic的路由信息只会包含Broker-a，然后消息发送者会向NameServer拉取最新的路由信息，此时就会消息发送者原本缓存了2个broker的路由信息，将会变为一个Broker的路由信息，则该Topic的消息永远不会发送到另外一个Broker，就出现了上述现象。
5. 一个新的消费组订阅一个已存在的Topic主题时，消费组是从该Topic的哪条消息开始消费呢？
RocketMQ实战：一个新的消费组初次启动时从何处开始消费呢？
看到这里，大家应该明白了，为什么设置的CONSUME_FROM_LAST_OFFSET，但消费组是从消息队列的开始处消费了吧，原因就是消息消费进度文件中并没有找到其消息消费进度，并且该队列在Broker端的最小偏移量为0，说的更直白点，consumequeue/topicName/queueNum的第一个消息消费队列文件为00000000000000000000,并且消息其对应的消息缓存在Broker端的内存中(pageCache)，其返回给消费端的偏移量为0，故会从0开始消费，而不是从队列的最大偏移量处开始消费。
