Browse Source

Startup sequence optimization

pull/6665/head
Andrii Shvaika 4 years ago
parent
commit
c5ff1252f5
  1. 39
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java

39
common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java

@ -90,20 +90,37 @@ public class HashPartitionService implements PartitionService {
@PostConstruct @PostConstruct
public void init() { public void init() {
this.hashFunction = forName(hashFunctionName); this.hashFunction = forName(hashFunctionName);
QueueKey coreKey = new QueueKey(ServiceType.TB_CORE);
partitionSizesMap.put(coreKey, corePartitions);
partitionTopicsMap.put(coreKey, coreTopic);
if (!isTransport(serviceInfoProvider.getServiceType())) {
doInitRuleEnginePartitions();
}
} }
@AfterStartUp(order = AfterStartUp.QUEUE_INFO_INITIALIZATION) @AfterStartUp(order = AfterStartUp.QUEUE_INFO_INITIALIZATION)
public void partitionsInit() { public void partitionsInit() {
QueueKey coreKey = new QueueKey(ServiceType.TB_CORE); if (isTransport(serviceInfoProvider.getServiceType())) {
partitionSizesMap.put(coreKey, corePartitions); doInitRuleEnginePartitions();
partitionTopicsMap.put(coreKey, coreTopic); }
}
List<QueueRoutingInfo> queueRoutingInfoList; private void doInitRuleEnginePartitions() {
List<QueueRoutingInfo> queueRoutingInfoList = getQueueRoutingInfos();
queueRoutingInfoList.forEach(queue -> {
QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queue);
partitionTopicsMap.put(queueKey, queue.getQueueTopic());
partitionSizesMap.put(queueKey, queue.getPartitions());
queuesById.put(queue.getQueueId(), queue);
});
}
private List<QueueRoutingInfo> getQueueRoutingInfos() {
List<QueueRoutingInfo> queueRoutingInfoList;
String serviceType = serviceInfoProvider.getServiceType(); String serviceType = serviceInfoProvider.getServiceType();
if (isTransport(serviceType)) {
if ("tb-transport".equals(serviceType)) {
//If transport started earlier than tb-core //If transport started earlier than tb-core
int getQueuesRetries = 10; int getQueuesRetries = 10;
while (true) { while (true) {
@ -128,13 +145,11 @@ public class HashPartitionService implements PartitionService {
} else { } else {
queueRoutingInfoList = queueRoutingInfoService.getAllQueuesRoutingInfo(); queueRoutingInfoList = queueRoutingInfoService.getAllQueuesRoutingInfo();
} }
return queueRoutingInfoList;
}
queueRoutingInfoList.forEach(queue -> { private boolean isTransport(String serviceType) {
QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queue); return "tb-transport".equals(serviceType);
partitionTopicsMap.put(queueKey, queue.getQueueTopic());
partitionSizesMap.put(queueKey, queue.getPartitions());
queuesById.put(queue.getQueueId(), queue);
});
} }
@Override @Override

Loading…
Cancel
Save