diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryStorage.java b/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryStorage.java index f51f5b8ae0..994ca26305 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryStorage.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryStorage.java @@ -23,16 +23,29 @@ import java.util.Collections; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @Slf4j public final class InMemoryStorage { private static InMemoryStorage instance; private final ConcurrentHashMap> storage; + private static ScheduledExecutorService statExecutor; private InMemoryStorage() { storage = new ConcurrentHashMap<>(); + statExecutor = Executors.newSingleThreadScheduledExecutor(); + statExecutor.scheduleAtFixedRate(this::printStats, 60, 60, TimeUnit.SECONDS); + } + + private void printStats() { + storage.forEach((topic, queue) -> { + if (queue.size() > 0) { + log.debug("Topic: [{}], Queue size: [{}]", topic, queue.size()); + } + }); } public static InMemoryStorage getInstance() { @@ -77,4 +90,9 @@ public final class InMemoryStorage { storage.clear(); } + public void destroy() { + if (statExecutor != null) { + statExecutor.shutdownNow(); + } + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueProducer.java b/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueProducer.java index cfcd788a16..84a9a1fdf0 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueProducer.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueProducer.java @@ -53,6 +53,6 @@ public class InMemoryTbQueueProducer implements TbQueuePro @Override public void stop() { - + storage.destroy(); } }