RocketMQ-broker狀態管理及資料統計
阿新 • • 發佈:2021-08-04
RocketMQ-broker狀態管理及資料統計
在RocketMQ中,狀態管理有BrokerStatsManager,ConsumerStatsManager,FilterServerStatsManager,其實現的方式都是一樣的。
這邊就拿BrokerStatsManager做介紹
一個StatsItemSet,代表著一項資料統計指標,這個指標定義了各個key的統計單項StatsItem,具體的值和次數使用原子類表示
BrokerStatsManager支援這些統一指標,同時每一個指標對應一個StateItemSet,裡面包含了各個topic和group的資訊
public staticfinal String TOPIC_PUT_NUMS = "TOPIC_PUT_NUMS"; public static final String TOPIC_PUT_SIZE = "TOPIC_PUT_SIZE"; public static final String GROUP_GET_NUMS = "GROUP_GET_NUMS"; public static final String GROUP_GET_SIZE = "GROUP_GET_SIZE"; public static final String SNDBCK_PUT_NUMS = "SNDBCK_PUT_NUMS";public static final String BROKER_PUT_NUMS = "BROKER_PUT_NUMS"; public static final String BROKER_GET_NUMS = "BROKER_GET_NUMS"; public static final String GROUP_GET_FROM_DISK_NUMS = "GROUP_GET_FROM_DISK_NUMS"; public static final String GROUP_GET_FROM_DISK_SIZE = "GROUP_GET_FROM_DISK_SIZE";public static final String BROKER_GET_FROM_DISK_NUMS = "BROKER_GET_FROM_DISK_NUMS"; public static final String BROKER_GET_FROM_DISK_SIZE = "BROKER_GET_FROM_DISK_SIZE"; // For commercial public static final String COMMERCIAL_SEND_TIMES = "COMMERCIAL_SEND_TIMES"; public static final String COMMERCIAL_SNDBCK_TIMES = "COMMERCIAL_SNDBCK_TIMES"; public static final String COMMERCIAL_RCV_TIMES = "COMMERCIAL_RCV_TIMES"; public static final String COMMERCIAL_RCV_EPOLLS = "COMMERCIAL_RCV_EPOLLS"; public static final String COMMERCIAL_SEND_SIZE = "COMMERCIAL_SEND_SIZE"; public static final String COMMERCIAL_RCV_SIZE = "COMMERCIAL_RCV_SIZE"; public static final String COMMERCIAL_PERM_FAILURES = "COMMERCIAL_PERM_FAILURES"; this.statsTable.put(TOPIC_PUT_NUMS, new StatsItemSet(TOPIC_PUT_NUMS, this.scheduledExecutorService, log)); this.statsTable.put(TOPIC_PUT_SIZE, new StatsItemSet(TOPIC_PUT_SIZE, this.scheduledExecutorService, log)); this.statsTable.put(GROUP_GET_NUMS, new StatsItemSet(GROUP_GET_NUMS, this.scheduledExecutorService, log)); this.statsTable.put(GROUP_GET_SIZE, new StatsItemSet(GROUP_GET_SIZE, this.scheduledExecutorService, log)); this.statsTable.put(GROUP_GET_LATENCY, new StatsItemSet(GROUP_GET_LATENCY, this.scheduledExecutorService, log)); this.statsTable.put(SNDBCK_PUT_NUMS, new StatsItemSet(SNDBCK_PUT_NUMS, this.scheduledExecutorService, log)); this.statsTable.put(BROKER_PUT_NUMS, new StatsItemSet(BROKER_PUT_NUMS, this.scheduledExecutorService, log)); this.statsTable.put(BROKER_GET_NUMS, new StatsItemSet(BROKER_GET_NUMS, this.scheduledExecutorService, log)); this.statsTable.put(GROUP_GET_FROM_DISK_NUMS, new StatsItemSet(GROUP_GET_FROM_DISK_NUMS, this.scheduledExecutorService, log)); this.statsTable.put(GROUP_GET_FROM_DISK_SIZE, new StatsItemSet(GROUP_GET_FROM_DISK_SIZE, this.scheduledExecutorService, log)); this.statsTable.put(BROKER_GET_FROM_DISK_NUMS, new StatsItemSet(BROKER_GET_FROM_DISK_NUMS, this.scheduledExecutorService, log)); this.statsTable.put(BROKER_GET_FROM_DISK_SIZE, new StatsItemSet(BROKER_GET_FROM_DISK_SIZE, this.scheduledExecutorService, log)); this.statsTable.put(COMMERCIAL_SEND_TIMES, new StatsItemSet(COMMERCIAL_SEND_TIMES, this.commercialExecutor, COMMERCIAL_LOG)); this.statsTable.put(COMMERCIAL_RCV_TIMES, new StatsItemSet(COMMERCIAL_RCV_TIMES, this.commercialExecutor, COMMERCIAL_LOG)); this.statsTable.put(COMMERCIAL_SEND_SIZE, new StatsItemSet(COMMERCIAL_SEND_SIZE, this.commercialExecutor, COMMERCIAL_LOG)); this.statsTable.put(COMMERCIAL_RCV_SIZE, new StatsItemSet(COMMERCIAL_RCV_SIZE, this.commercialExecutor, COMMERCIAL_LOG)); this.statsTable.put(COMMERCIAL_RCV_EPOLLS, new StatsItemSet(COMMERCIAL_RCV_EPOLLS, this.commercialExecutor, COMMERCIAL_LOG)); this.statsTable.put(COMMERCIAL_SNDBCK_TIMES, new StatsItemSet(COMMERCIAL_SNDBCK_TIMES, this.commercialExecutor, COMMERCIAL_LOG)); this.statsTable.put(COMMERCIAL_PERM_FAILURES, new StatsItemSet(COMMERCIAL_PERM_FAILURES, this.commercialExecutor, COMMERCIAL_LOG));
最後這些指標都會儲存在 private final HashMap<String, StatsItemSet> statsTable = new HashMap<String, StatsItemSet>();
一個指標StatsItemSet裡面有如下屬性
private final ConcurrentMap<String/* key */, StatsItem> statsItemTable = new ConcurrentHashMap<String, StatsItem>(128); private final String statsName; private final ScheduledExecutorService scheduledExecutorService; private final Logger log;
比如一個TOPIC_PUT_NUMS的指標,在其statsItemTable 包含了各個topic的message數量和存放次數。其中key就是topic。
然後會初始化一些定時任務,比如會在每隔10s把當前的狀態封裝之後放在csListMinute,最多放6個統計記錄,然後在一分鐘內列印結果,格式如下
"TOPIC_PUT_NUMS [topicname] Stats In One Minute, SUM: %d TPS: %.2f AVGPT: %.2
再次之前會進行簡單計算程式碼如下
private static StatsSnapshot computeStatsData(final LinkedList<CallSnapshot> csList) { StatsSnapshot statsSnapshot = new StatsSnapshot(); synchronized (csList) { double tps = 0; double avgpt = 0; long sum = 0; if (!csList.isEmpty()) { CallSnapshot first = csList.getFirst(); CallSnapshot last = csList.getLast(); // 比如統計前後的值差距 sum = last.getValue() - first.getValue(); // last.getTimestamp() - first.getTimestamp()代表的是統計的時候時間差 tps = (sum * 1000.0d) / (last.getTimestamp() - first.getTimestamp()); // 次數差距 long timesDiff = last.getTimes() - first.getTimes(); if (timesDiff > 0) { // 平均每次的達到的數量級 ,在TOPIC_PUT_NUMS中表示 每次在在該topic中平均寫入avgpt個message avgpt = (sum * 1.0d) / timesDiff; } } statsSnapshot.setSum(sum); statsSnapshot.setTps(tps); statsSnapshot.setAvgpt(avgpt); } return statsSnapshot; }
每隔一小時,一天的統計結果處理邏輯雷同。