|
|
@ -18,6 +18,7 @@ package org.thingsboard.server.queue.discovery; |
|
|
import com.google.common.hash.HashCode; |
|
|
import com.google.common.hash.HashCode; |
|
|
import com.google.common.hash.HashFunction; |
|
|
import com.google.common.hash.HashFunction; |
|
|
import com.google.common.hash.Hashing; |
|
|
import com.google.common.hash.Hashing; |
|
|
|
|
|
import lombok.Getter; |
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
import org.springframework.beans.factory.annotation.Value; |
|
|
import org.springframework.beans.factory.annotation.Value; |
|
|
import org.springframework.context.ApplicationEventPublisher; |
|
|
import org.springframework.context.ApplicationEventPublisher; |
|
|
@ -35,6 +36,7 @@ import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; |
|
|
import javax.annotation.PostConstruct; |
|
|
import javax.annotation.PostConstruct; |
|
|
import java.nio.charset.StandardCharsets; |
|
|
import java.nio.charset.StandardCharsets; |
|
|
import java.util.ArrayList; |
|
|
import java.util.ArrayList; |
|
|
|
|
|
import java.util.Comparator; |
|
|
import java.util.HashMap; |
|
|
import java.util.HashMap; |
|
|
import java.util.HashSet; |
|
|
import java.util.HashSet; |
|
|
import java.util.List; |
|
|
import java.util.List; |
|
|
@ -48,7 +50,7 @@ import java.util.stream.Collectors; |
|
|
|
|
|
|
|
|
@Service |
|
|
@Service |
|
|
@Slf4j |
|
|
@Slf4j |
|
|
public class ConsistentHashPartitionService implements PartitionService { |
|
|
public class HashPartitionService implements PartitionService { |
|
|
|
|
|
|
|
|
@Value("${queue.core.topic}") |
|
|
@Value("${queue.core.topic}") |
|
|
private String coreTopic; |
|
|
private String coreTopic; |
|
|
@ -56,8 +58,6 @@ public class ConsistentHashPartitionService implements PartitionService { |
|
|
private Integer corePartitions; |
|
|
private Integer corePartitions; |
|
|
@Value("${queue.partitions.hash_function_name:murmur3_128}") |
|
|
@Value("${queue.partitions.hash_function_name:murmur3_128}") |
|
|
private String hashFunctionName; |
|
|
private String hashFunctionName; |
|
|
@Value("${queue.partitions.virtual_nodes_size:16}") |
|
|
|
|
|
private Integer virtualNodesSize; |
|
|
|
|
|
|
|
|
|
|
|
private final ApplicationEventPublisher applicationEventPublisher; |
|
|
private final ApplicationEventPublisher applicationEventPublisher; |
|
|
private final TbServiceInfoProvider serviceInfoProvider; |
|
|
private final TbServiceInfoProvider serviceInfoProvider; |
|
|
@ -76,10 +76,10 @@ public class ConsistentHashPartitionService implements PartitionService { |
|
|
|
|
|
|
|
|
private HashFunction hashFunction; |
|
|
private HashFunction hashFunction; |
|
|
|
|
|
|
|
|
public ConsistentHashPartitionService(TbServiceInfoProvider serviceInfoProvider, |
|
|
public HashPartitionService(TbServiceInfoProvider serviceInfoProvider, |
|
|
TenantRoutingInfoService tenantRoutingInfoService, |
|
|
TenantRoutingInfoService tenantRoutingInfoService, |
|
|
ApplicationEventPublisher applicationEventPublisher, |
|
|
ApplicationEventPublisher applicationEventPublisher, |
|
|
TbQueueRuleEngineSettings tbQueueRuleEngineSettings) { |
|
|
TbQueueRuleEngineSettings tbQueueRuleEngineSettings) { |
|
|
this.serviceInfoProvider = serviceInfoProvider; |
|
|
this.serviceInfoProvider = serviceInfoProvider; |
|
|
this.tenantRoutingInfoService = tenantRoutingInfoService; |
|
|
this.tenantRoutingInfoService = tenantRoutingInfoService; |
|
|
this.applicationEventPublisher = applicationEventPublisher; |
|
|
this.applicationEventPublisher = applicationEventPublisher; |
|
|
@ -128,20 +128,22 @@ public class ConsistentHashPartitionService implements PartitionService { |
|
|
public void recalculatePartitions(ServiceInfo currentService, List<ServiceInfo> otherServices) { |
|
|
public void recalculatePartitions(ServiceInfo currentService, List<ServiceInfo> otherServices) { |
|
|
logServiceInfo(currentService); |
|
|
logServiceInfo(currentService); |
|
|
otherServices.forEach(this::logServiceInfo); |
|
|
otherServices.forEach(this::logServiceInfo); |
|
|
Map<ServiceQueueKey, ConsistentHashCircle<ServiceInfo>> circles = new HashMap<>(); |
|
|
Map<ServiceQueueKey, List<ServiceInfo>> queueServicesMap = new HashMap<>(); |
|
|
addNode(circles, currentService); |
|
|
addNode(queueServicesMap, currentService); |
|
|
for (ServiceInfo other : otherServices) { |
|
|
for (ServiceInfo other : otherServices) { |
|
|
addNode(circles, other); |
|
|
addNode(queueServicesMap, other); |
|
|
} |
|
|
} |
|
|
|
|
|
queueServicesMap.values().forEach(list -> list.sort((a, b) -> a.getServiceId().compareTo(b.getServiceId()))); |
|
|
|
|
|
|
|
|
ConcurrentMap<ServiceQueueKey, List<Integer>> oldPartitions = myPartitions; |
|
|
ConcurrentMap<ServiceQueueKey, List<Integer>> oldPartitions = myPartitions; |
|
|
TenantId myIsolatedOrSystemTenantId = getSystemOrIsolatedTenantId(currentService); |
|
|
TenantId myIsolatedOrSystemTenantId = getSystemOrIsolatedTenantId(currentService); |
|
|
myPartitions = new ConcurrentHashMap<>(); |
|
|
myPartitions = new ConcurrentHashMap<>(); |
|
|
partitionSizes.forEach((type, size) -> { |
|
|
partitionSizes.forEach((serviceQueue, size) -> { |
|
|
ServiceQueueKey myServiceQueueKey = new ServiceQueueKey(type, myIsolatedOrSystemTenantId); |
|
|
ServiceQueueKey myServiceQueueKey = new ServiceQueueKey(serviceQueue, myIsolatedOrSystemTenantId); |
|
|
for (int i = 0; i < size; i++) { |
|
|
for (int i = 0; i < size; i++) { |
|
|
ServiceInfo serviceInfo = resolveByPartitionIdx(circles.get(myServiceQueueKey), i); |
|
|
ServiceInfo serviceInfo = resolveByPartitionIdx(queueServicesMap.get(myServiceQueueKey), i); |
|
|
if (currentService.equals(serviceInfo)) { |
|
|
if (currentService.equals(serviceInfo)) { |
|
|
ServiceQueueKey serviceQueueKey = new ServiceQueueKey(type, getSystemOrIsolatedTenantId(serviceInfo)); |
|
|
ServiceQueueKey serviceQueueKey = new ServiceQueueKey(serviceQueue, getSystemOrIsolatedTenantId(serviceInfo)); |
|
|
myPartitions.computeIfAbsent(serviceQueueKey, key -> new ArrayList<>()).add(i); |
|
|
myPartitions.computeIfAbsent(serviceQueueKey, key -> new ArrayList<>()).add(i); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
@ -293,7 +295,7 @@ public class ConsistentHashPartitionService implements PartitionService { |
|
|
return new TenantId(new UUID(serviceInfo.getTenantIdMSB(), serviceInfo.getTenantIdLSB())); |
|
|
return new TenantId(new UUID(serviceInfo.getTenantIdMSB(), serviceInfo.getTenantIdLSB())); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
private void addNode(Map<ServiceQueueKey, ConsistentHashCircle<ServiceInfo>> circles, ServiceInfo instance) { |
|
|
private void addNode(Map<ServiceQueueKey, List<ServiceInfo>> queueServiceList, ServiceInfo instance) { |
|
|
TenantId tenantId = getSystemOrIsolatedTenantId(instance); |
|
|
TenantId tenantId = getSystemOrIsolatedTenantId(instance); |
|
|
for (String serviceTypeStr : instance.getServiceTypesList()) { |
|
|
for (String serviceTypeStr : instance.getServiceTypesList()) { |
|
|
ServiceType serviceType = ServiceType.valueOf(serviceTypeStr.toUpperCase()); |
|
|
ServiceType serviceType = ServiceType.valueOf(serviceTypeStr.toUpperCase()); |
|
|
@ -302,34 +304,20 @@ public class ConsistentHashPartitionService implements PartitionService { |
|
|
ServiceQueueKey serviceQueueKey = new ServiceQueueKey(new ServiceQueue(serviceType, queue.getName()), tenantId); |
|
|
ServiceQueueKey serviceQueueKey = new ServiceQueueKey(new ServiceQueue(serviceType, queue.getName()), tenantId); |
|
|
partitionSizes.put(new ServiceQueue(ServiceType.TB_RULE_ENGINE, queue.getName()), queue.getPartitions()); |
|
|
partitionSizes.put(new ServiceQueue(ServiceType.TB_RULE_ENGINE, queue.getName()), queue.getPartitions()); |
|
|
partitionTopics.put(new ServiceQueue(ServiceType.TB_RULE_ENGINE, queue.getName()), queue.getTopic()); |
|
|
partitionTopics.put(new ServiceQueue(ServiceType.TB_RULE_ENGINE, queue.getName()), queue.getTopic()); |
|
|
for (int i = 0; i < virtualNodesSize; i++) { |
|
|
queueServiceList.computeIfAbsent(serviceQueueKey, key -> new ArrayList<>()).add(instance); |
|
|
circles.computeIfAbsent(serviceQueueKey, key -> new ConsistentHashCircle<>()).put(hash(instance, i).asLong(), instance); |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
} |
|
|
} else { |
|
|
} else { |
|
|
ServiceQueueKey serviceQueueKey = new ServiceQueueKey(new ServiceQueue(serviceType), tenantId); |
|
|
ServiceQueueKey serviceQueueKey = new ServiceQueueKey(new ServiceQueue(serviceType), tenantId); |
|
|
for (int i = 0; i < virtualNodesSize; i++) { |
|
|
queueServiceList.computeIfAbsent(serviceQueueKey, key -> new ArrayList<>()).add(instance); |
|
|
circles.computeIfAbsent(serviceQueueKey, key -> new ConsistentHashCircle<>()).put(hash(instance, i).asLong(), instance); |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
private ServiceInfo resolveByPartitionIdx(ConsistentHashCircle<ServiceInfo> circle, Integer partitionIdx) { |
|
|
private ServiceInfo resolveByPartitionIdx(List<ServiceInfo> servers, Integer partitionIdx) { |
|
|
if (circle == null || circle.isEmpty()) { |
|
|
if (servers == null || servers.isEmpty()) { |
|
|
return null; |
|
|
return null; |
|
|
} |
|
|
} |
|
|
Long hash = hashFunction.newHasher().putInt(partitionIdx).hash().asLong(); |
|
|
return servers.get(partitionIdx % servers.size()); |
|
|
if (!circle.containsKey(hash)) { |
|
|
|
|
|
ConcurrentNavigableMap<Long, ServiceInfo> tailMap = circle.tailMap(hash); |
|
|
|
|
|
hash = tailMap.isEmpty() ? |
|
|
|
|
|
circle.firstKey() : tailMap.firstKey(); |
|
|
|
|
|
} |
|
|
|
|
|
return circle.get(hash); |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
private HashCode hash(ServiceInfo instance, int i) { |
|
|
|
|
|
return hashFunction.newHasher().putString(instance.getServiceId(), StandardCharsets.UTF_8).putInt(i).hash(); |
|
|
|
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
public static HashFunction forName(String name) { |
|
|
public static HashFunction forName(String name) { |
|
|
@ -338,12 +326,11 @@ public class ConsistentHashPartitionService implements PartitionService { |
|
|
return Hashing.murmur3_32(); |
|
|
return Hashing.murmur3_32(); |
|
|
case "murmur3_128": |
|
|
case "murmur3_128": |
|
|
return Hashing.murmur3_128(); |
|
|
return Hashing.murmur3_128(); |
|
|
case "crc32": |
|
|
case "sha256": |
|
|
return Hashing.crc32(); |
|
|
return Hashing.sha256(); |
|
|
case "md5": |
|
|
|
|
|
return Hashing.md5(); |
|
|
|
|
|
default: |
|
|
default: |
|
|
throw new IllegalArgumentException("Can't find hash function with name " + name); |
|
|
throw new IllegalArgumentException("Can't find hash function with name " + name); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
} |
|
|
} |
Volodymyr Babak
6 years ago