开箱即用！动态修改RocketMQ线程池

@Service@RocketMQMessageListener(topic="test-topic", consumerGroup="my-consumer_test-topic", consumeThreadNumber=20)
publicclassMessageConsumeimplementsRocketMQListener {
@OverridepublicvoidonMessage(Stringmessage) {
log.info(Thread.currentThread().getName());
log.info("Message: {}", message);
    }
}
@Service@RocketMQMessageListener(topic="test-topic", consumerGroup="my-consumer_test-topic_1", consumeThreadNumber=20)
publicclassMessageConsumeimplementsRocketMQListener {
@OverridepublicvoidonMessage(Stringmessage) {
log.info(Thread.currentThread().getName());
log.info("Message: {}", message);
    }
}
@Service@RocketMQMessageListener(topic="test-topic", consumerGroup="my-consumer_test-topic_2", consumeThreadNumber=20)
publicclassMessageConsumeimplementsRocketMQListener {
@OverridepublicvoidonMessage(Stringmessage) {
log.info(Thread.currentThread().getName());
log.info("Message: {}", message);
    }
}

@GetMapping("/message/send")
publicStringsendMessage() {
IntStream.range(0, 1000).forEach(i-> {
rocketMQTemplate.convertAndSend("test-topic", newDate().toString());
    });
return"success";
}

@OverridepublicvoidonApplicationEvent(ApplicationStartedEventevent) {
MapcontainerMap=ApplicationContextHolder.getBeansOfType(DefaultRocketMQListenerContainer.class);
try {
for (DefaultRocketMQListenerContainercontainer : containerMap.values()) {
DefaultMQPushConsumerdefaultMQPushConsumer=container.getConsumer();
if (defaultMQPushConsumer!=null) {
ConsumeMessageServiceconsumeMessageService=defaultMQPushConsumer.getDefaultMQPushConsumerImpl().getConsumeMessageService();
ThreadPoolExecutorconsumeExecutor= (ThreadPoolExecutor) ReflectUtil.getFieldValue(consumeMessageService, "consumeExecutor");
rocketmqConsumeExecutor.put(container.getConsumerGroup(), consumeExecutor);
            }
        }
    } catch (Exceptionex) {
log.error("Failed to get RocketMQ thread pool.", ex);
    }
}

publicDiscoveryClient(HttpAgenthttpAgent, InstanceInfoinstanceInfo) {
    ...
register();
}
booleanregister() {
log.info("{}{} - registering service...", PREFIX, appPathIdentifier);
StringurlPath=BASE_PATH+"/apps/register/";
ResultregisterResult;
try {
registerResult=httpAgent.httpPostByDiscovery(urlPath, instanceInfo);
    } catch (Exceptionex) {
registerResult=Results.failure(ErrorCodeEnum.SERVICE_ERROR);
log.error("{}{} - registration failed: {}", PREFIX, appPathIdentifier, ex.getMessage());
    }
if (log.isInfoEnabled()) {
log.info("{}{} - registration status: {}", PREFIX, appPathIdentifier, registerResult.isSuccess() ?"success" : "fail");
    }
returnregisterResult.isSuccess();
}

publicDiscoveryClient(HttpAgenthttpAgent, InstanceInfoinstanceInfo) {
// ...initScheduledTasks();
}
// 定时任务线程池privatevoidinitScheduledTasks() {
scheduler.scheduleWithFixedDelay(newHeartbeatThread(), 30, 30, TimeUnit.SECONDS);
}
// 心跳线程publicclassHeartbeatThreadimplementsRunnable {
@Overridepublicvoidrun() {
if (renew()) {
lastSuccessfulHeartbeatTimestamp=System.currentTimeMillis();
        }
    }
}
privatebooleanrenew() {
ResultrenewResult;
try {
InstanceInfo.InstanceRenewinstanceRenew=newInstanceInfo.InstanceRenew()
                .setAppName(instanceInfo.getAppName())
                .setInstanceId(instanceInfo.getInstanceId())
                .setLastDirtyTimestamp(instanceInfo.getLastDirtyTimestamp().toString())
                .setStatus(instanceInfo.getStatus().toString());
// 检查实例信息是否正常renewResult=httpAgent.httpPostByDiscovery(BASE_PATH+"/apps/renew", instanceRenew);
if (Objects.equals(ErrorCodeEnum.NOT_FOUND.getCode(), renewResult.getCode())) {
longtimestamp=instanceInfo.setIsDirtyWithTime();
booleansuccess=register();
ThreadPoolAdapterRegisteradapterRegister=ApplicationContextHolder.getBean(ThreadPoolAdapterRegister.class);
adapterRegister.register();
if (success) {
instanceInfo.unsetIsDirty(timestamp);
            }
returnsuccess;
        }
returnrenewResult.isSuccess();
    } catch (Exceptionex) {
log.error(PREFIX+"{} - was unable to send heartbeat!", appPathIdentifier, ex);
returnfalse;
    }
}
// 注册实例信息publicvoidregister() {
MapthreadPoolAdapterMap=ApplicationContextHolder.getBeansOfType(ThreadPoolAdapter.class);
ListthreadPoolAdapterCacheConfigs=getThreadPoolAdapterCacheConfigs(threadPoolAdapterMap);
doRegister(threadPoolAdapterCacheConfigs);
}

@OverridepublicbooleanupdateThreadPool(ThreadPoolAdapterParameterthreadPoolAdapterParameter) {
StringthreadPoolKey=threadPoolAdapterParameter.getThreadPoolKey();
// 通过threadPoolKey拿到执行器ThreadPoolExecutorrocketMQConsumeExecutor=rocketmqConsumeExecutor.get(threadPoolKey);
if (rocketMQConsumeExecutor!=null) {
intoriginalCoreSize=rocketMQConsumeExecutor.getCorePoolSize();
intoriginalMaximumPoolSize=rocketMQConsumeExecutor.getMaximumPoolSize();
// 更新核心线程数和最大线程数rocketMQConsumeExecutor.setCorePoolSize(threadPoolAdapterParameter.getCorePoolSize());
rocketMQConsumeExecutor.setMaximumPoolSize(threadPoolAdapterParameter.getMaximumPoolSize());
log.info("[{}] RocketMQ consumption thread pool parameter change. coreSize: {}, maximumSize: {}",
threadPoolKey,
String.format(CHANGE_DELIMITER, originalCoreSize, rocketMQConsumeExecutor.getCorePoolSize()),
String.format(CHANGE_DELIMITER, originalMaximumPoolSize, rocketMQConsumeExecutor.getMaximumPoolSize()));
returntrue;
    }
log.warn("[{}] RocketMQ consuming thread pool not found.", threadPoolKey);
returnfalse;
}

2022-12-2615:35:35.472INFO9726--- [nio-8099-exec-2] c.h.s.s.c.ThreadPoolAdapterController    : [RocketMQ] Changethird-partythreadpooldata. key: my-consumer_test-topic, coreSize: 100, maximumSize: 1002022-12-2615:35:35.473INFO9726--- [nio-8099-exec-2] c.h.a.r.RocketMQThreadPoolAdapter        : [my-consumer_test-topic] RocketMQconsumptionthreadpoolparameterchange. coreSize: 20=>100, maximumSize: 20=>100