@Override
    public void run() {
        // Indicator output every 30 seconds
        while (isStarted.get()) {
            try {
                for (Map.Entry<String, AtomicLong> entry : counterMap.entrySet()) {
                    long currCount = entry.getValue().get();
                    AtomicLong befCount = befCountMap.get(entry.getKey());
                    if (befCount == null) {
                        AtomicLong tmpCount = new AtomicLong(0);
                        befCount = befCountMap.putIfAbsent(entry.getKey(), tmpCount);
                        if (befCount == null) {
                            befCount = tmpCount;
                        }
                    }
                    if (isProducer) {
                        logger.info("********* Current {} Message sent count is {}, dlt is {}",
                                entry.getKey(), currCount, (currCount - befCount.get()));
                    } else {
                        logger.info("********* Current {} Message received count is {}, dlt is {}",
                                entry.getKey(), currCount, (currCount - befCount.get()));
                    }
                    befCountMap.get(entry.getKey()).set(currCount);
                }
            } catch (Throwable t) {
                // ignore
            }
            ThreadUtils.sleep(30000);
        }
    }
This demo shows how to collect and report message received statistics.
inlong-tubemq/tubemq-example/src/main/java/org/apache/inlong/tubemq/example/MsgSendReceiveStats.java