Browse Source

using repartitionExecutor instead of synchronized

pull/6534/head
YevhenBondarenko 4 years ago
parent
commit
4f7c166566
  1. 6
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java
  2. 3
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java

6
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java

@ -393,10 +393,10 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
tbDeviceRpcService.processRpcResponseFromDevice(response);
callback.onSuccess();
} else if (nfMsg.hasQueueUpdateMsg()) {
updateQueue(nfMsg.getQueueUpdateMsg());
repartitionExecutor.execute(() -> updateQueue(nfMsg.getQueueUpdateMsg()));
callback.onSuccess();
} else if (nfMsg.hasQueueDeleteMsg()) {
deleteQueue(nfMsg.getQueueDeleteMsg());
repartitionExecutor.execute(() -> deleteQueue(nfMsg.getQueueDeleteMsg()));
callback.onSuccess();
} else {
log.trace("Received notification with missing handler");
@ -404,7 +404,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
}
}
private synchronized void updateQueue(TransportProtos.QueueUpdateMsg queueUpdateMsg) {
private void updateQueue(TransportProtos.QueueUpdateMsg queueUpdateMsg) {
log.info("Received queue update msg: [{}]", queueUpdateMsg);
String queueName = queueUpdateMsg.getQueueName();
TenantId tenantId = new TenantId(new UUID(queueUpdateMsg.getTenantIdMSB(), queueUpdateMsg.getTenantIdLSB()));

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

@ -173,7 +173,6 @@ public class HashPartitionService implements PartitionService {
} else {
QueueRoutingInfo queueRoutingInfo = queuesById.get(queueId);
//TODO: replace if we can notify CheckPoint rule nodes about queue changes
if (queueRoutingInfo == null) {
log.debug("Queue was removed but still used in CheckPoint rule node. [{}][{}]", tenantId, entityId);
queueKey = getMainQueueKey(serviceType, tenantId);
@ -205,6 +204,8 @@ public class HashPartitionService implements PartitionService {
@Override
public synchronized void recalculatePartitions(ServiceInfo currentService, List<ServiceInfo> otherServices) {
partitionsInit();
tbTransportServicesByType.clear();
logServiceInfo(currentService);
otherServices.forEach(this::logServiceInfo);

Loading…
Cancel
Save