@Override
    public void add(String statsKey, long msgCnt, long msgSize) {
        // Increment write reference count
        WritableUnit selectedUnit = switchableUnits[getIndex()];
        selectedUnit.refCnt.incValue();
        try {
            // Accumulate statistics information
            ConcurrentHashMap<String, TrafficStatsUnit> tmpStatsSetMap =
                    selectedUnit.statsUnitMap;
            TrafficStatsUnit trafficStatsSet = tmpStatsSetMap.get(statsKey);
            if (trafficStatsSet == null) {
                TrafficStatsUnit tmpStatsSet = new TrafficStatsUnit("msg_cnt", "msg_size", null);
                trafficStatsSet = tmpStatsSetMap.putIfAbsent(statsKey, tmpStatsSet);
                if (trafficStatsSet == null) {
                    trafficStatsSet = tmpStatsSet;
                }
            }
            trafficStatsSet.addMsgCntAndSize(msgCnt, msgSize);
        } finally {
            // Decrement write reference count
            selectedUnit.refCnt.decValue();
        }
    }
Initial traffic statistics service

@param logFileName      the output file name
@param countType        the statistic type
@param scanIntervalMs   the snapshot interval
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/stats/TrafficStatsService.java