diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index dbd58d540c..ef1606a3a3 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java @@ -60,7 +60,6 @@ import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.util.ArrayList; import java.util.Collections; -import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.List; @@ -68,6 +67,7 @@ import java.util.Map; import java.util.Optional; import java.util.Set; import java.util.TreeMap; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.function.Consumer; @@ -120,8 +120,8 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio } } - private final Map> subscriptionsByEntityId = new HashMap<>(); - private final Map> subscriptionsByWsSessionId = new HashMap<>(); + private final Map> subscriptionsByEntityId = new ConcurrentHashMap<>(); + private final Map> subscriptionsByWsSessionId = new ConcurrentHashMap<>(); @Override public void addLocalWsSubscription(String sessionId, EntityId entityId, SubscriptionState sub) { @@ -453,9 +453,9 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio } private void registerSubscription(String sessionId, EntityId entityId, Subscription subscription) { - Set deviceSubscriptions = subscriptionsByEntityId.computeIfAbsent(entityId, k -> new HashSet<>()); + Set deviceSubscriptions = subscriptionsByEntityId.computeIfAbsent(entityId, k -> ConcurrentHashMap.newKeySet()); deviceSubscriptions.add(subscription); - Map sessionSubscriptions = subscriptionsByWsSessionId.computeIfAbsent(sessionId, k -> new HashMap<>()); + Map sessionSubscriptions = subscriptionsByWsSessionId.computeIfAbsent(sessionId, k -> new ConcurrentHashMap<>()); sessionSubscriptions.put(subscription.getSubscriptionId(), subscription); }