@Override
    public void add(Map<String, TrafficInfo> trafficInfos) {
        TrafficStatsUnit tmpStatsSet;
        TrafficStatsUnit trafficStatsSet;
        // Increment write reference count
        WritableUnit selectedUnit = switchableUnits[getIndex()];
        selectedUnit.refCnt.incValue();
        try {
            // Accumulate statistics information
            ConcurrentHashMap<String, TrafficStatsUnit> tmpStatsSetMap =
                    selectedUnit.statsUnitMap;
            for (Entry<String, TrafficInfo> entry : trafficInfos.entrySet()) {
                trafficStatsSet = tmpStatsSetMap.get(entry.getKey());
                if (trafficStatsSet == null) {
                    tmpStatsSet = new TrafficStatsUnit("msg_cnt", "msg_size", null);
                    trafficStatsSet = tmpStatsSetMap.putIfAbsent(entry.getKey(), tmpStatsSet);
                    if (trafficStatsSet == null) {
                        trafficStatsSet = tmpStatsSet;
                    }
                }
                trafficStatsSet.addMsgCntAndSize(
                        entry.getValue().getMsgCount(), entry.getValue().getMsgSize());
            }
        } 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