mirror of
https://github.com/clash-verge-rev/clash-verge-rev.git
synced 2026-10-30 02:32:09 +08:00
refactor(traffic-monitor): Move traffic sampling/compression to a dedicated Web Worker (#5588)
* refactor(traffic-monitor): Move traffic sampling/compression to a dedicated Web Worker * refactor(traffic-monitor): introduce shared sampler, safe fallback, and decoupled range handling * refactor(traffic-monitor): add optional subscribe flag and fix range handling * fix(traffic-monitor): age out idle samples and prevent ref-count updates after unmount
This commit is contained in:
Sline
@@ -110,7 +110,7 @@ export const EnhancedCanvasTrafficGraph = memo(
|
||||
const { t } = useTranslation();
|
||||
|
||||
// 使用增强版全局流量数据管理
|
||||
const { dataPoints, getDataForTimeRange, samplerStats } =
|
||||
const { dataPoints, requestRange, samplerStats } =
|
||||
useTrafficGraphDataEnhanced();
|
||||
|
||||
// 基础状态
|
||||
@@ -183,8 +183,7 @@ export const EnhancedCanvasTrafficGraph = memo(
|
||||
|
||||
// 监听数据变化
|
||||
useEffect(() => {
|
||||
const timeRangeData = getDataForTimeRange(timeRange);
|
||||
updateDisplayData(timeRangeData);
|
||||
updateDisplayData(dataPoints);
|
||||
|
||||
return () => {
|
||||
if (debounceTimeoutRef.current !== null) {
|
||||
@@ -192,7 +191,11 @@ export const EnhancedCanvasTrafficGraph = memo(
|
||||
debounceTimeoutRef.current = null;
|
||||
}
|
||||
};
|
||||
}, [dataPoints, timeRange, getDataForTimeRange, updateDisplayData]);
|
||||
}, [dataPoints, updateDisplayData]);
|
||||
|
||||
useEffect(() => {
|
||||
requestRange(timeRange);
|
||||
}, [requestRange, timeRange]);
|
||||
|
||||
useEffect(() => {
|
||||
if (displayData.length === 0) {
|
||||
|
||||
@@ -15,7 +15,7 @@ export const useTrafficData = () => {
|
||||
const trafficRef = useRef<TrafficRef>(null);
|
||||
const {
|
||||
graphData: { appendData },
|
||||
} = useTrafficMonitorEnhanced();
|
||||
} = useTrafficMonitorEnhanced({ subscribe: false });
|
||||
const ws = useRef<MihomoWebSocket | null>(null);
|
||||
const wsFirstConnection = useRef<boolean>(true);
|
||||
const timeoutRef = useRef<ReturnType<typeof setTimeout>>(null);
|
||||
|
||||
+393
-223
@@ -1,61 +1,51 @@
|
||||
import { useEffect, useRef, useCallback, useReducer } from "react";
|
||||
import {
|
||||
useCallback,
|
||||
useEffect,
|
||||
useMemo,
|
||||
useReducer,
|
||||
useRef,
|
||||
useState,
|
||||
} from "react";
|
||||
import { Traffic } from "tauri-plugin-mihomo-api";
|
||||
|
||||
import type {
|
||||
ISamplerStats,
|
||||
ITrafficDataPoint,
|
||||
ITrafficWorkerSnapshotMessage,
|
||||
TrafficWorkerRequestMessage,
|
||||
TrafficWorkerResponseMessage,
|
||||
} from "@/types/traffic-monitor";
|
||||
import { debugLog } from "@/utils/debug";
|
||||
import { TrafficDataSampler, formatTrafficName } from "@/utils/traffic-sampler";
|
||||
|
||||
// 增强的流量数据点接口
|
||||
export interface ITrafficDataPoint {
|
||||
up: number;
|
||||
down: number;
|
||||
timestamp: number;
|
||||
name: string;
|
||||
}
|
||||
|
||||
// 压缩的数据点（用于长期存储）
|
||||
interface ICompressedDataPoint {
|
||||
up: number;
|
||||
down: number;
|
||||
timestamp: number;
|
||||
samples: number; // 压缩了多少个原始数据点
|
||||
}
|
||||
|
||||
// 数据采样器配置
|
||||
interface ISamplingConfig {
|
||||
// 原始数据保持时间（分钟）
|
||||
rawDataMinutes: number;
|
||||
// 压缩数据保持时间（分钟）
|
||||
compressedDataMinutes: number;
|
||||
// 压缩比例（多少个原始点压缩成1个）
|
||||
compressionRatio: number;
|
||||
}
|
||||
export type { ITrafficDataPoint } from "@/types/traffic-monitor";
|
||||
|
||||
// 引用计数管理器
|
||||
class ReferenceCounter {
|
||||
private count = 0;
|
||||
private callbacks: (() => void)[] = [];
|
||||
private callbacks = new Set<() => void>();
|
||||
|
||||
private notify() {
|
||||
this.callbacks.forEach((cb) => cb());
|
||||
}
|
||||
|
||||
increment(): () => void {
|
||||
this.count++;
|
||||
debugLog(`[ReferenceCounter] 引用计数增加: ${this.count}`);
|
||||
|
||||
if (this.count === 1) {
|
||||
// 从0到1，开始数据收集
|
||||
this.callbacks.forEach((cb) => cb());
|
||||
}
|
||||
this.notify();
|
||||
|
||||
return () => {
|
||||
this.count--;
|
||||
debugLog(`[ReferenceCounter] 引用计数减少: ${this.count}`);
|
||||
|
||||
if (this.count === 0) {
|
||||
// 从1到0，停止数据收集
|
||||
this.callbacks.forEach((cb) => cb());
|
||||
}
|
||||
this.notify();
|
||||
};
|
||||
}
|
||||
|
||||
onCountChange(callback: () => void) {
|
||||
this.callbacks.push(callback);
|
||||
this.callbacks.add(callback);
|
||||
return () => this.callbacks.delete(callback);
|
||||
}
|
||||
|
||||
getCount(): number {
|
||||
@@ -63,225 +53,405 @@ class ReferenceCounter {
|
||||
}
|
||||
}
|
||||
|
||||
// 智能数据采样器
|
||||
class TrafficDataSampler {
|
||||
private rawBuffer: ITrafficDataPoint[] = [];
|
||||
private compressedBuffer: ICompressedDataPoint[] = [];
|
||||
private config: ISamplingConfig;
|
||||
private compressionQueue: ITrafficDataPoint[] = [];
|
||||
const WORKER_CONFIG = {
|
||||
rawDataMinutes: 10,
|
||||
compressedDataMinutes: 60,
|
||||
compressionRatio: 5,
|
||||
snapshotIntervalMs: 250,
|
||||
defaultRangeMinutes: 10,
|
||||
};
|
||||
|
||||
constructor(config: ISamplingConfig) {
|
||||
this.config = config;
|
||||
}
|
||||
class InlineTrafficMonitor {
|
||||
private config = { ...WORKER_CONFIG };
|
||||
private sampler = new TrafficDataSampler(this.config);
|
||||
private throttleTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
private currentRange = this.config.defaultRangeMinutes;
|
||||
private lastTimestamp: number | undefined;
|
||||
|
||||
addDataPoint(point: ITrafficDataPoint): void {
|
||||
// 添加到原始缓冲区
|
||||
this.rawBuffer.push(point);
|
||||
constructor(
|
||||
private emit: (snapshot: ITrafficWorkerSnapshotMessage) => void,
|
||||
) {}
|
||||
|
||||
// 清理过期的原始数据
|
||||
const rawCutoff = Date.now() - this.config.rawDataMinutes * 60 * 1000;
|
||||
this.rawBuffer = this.rawBuffer.filter((p) => p.timestamp > rawCutoff);
|
||||
|
||||
// 添加到压缩队列
|
||||
this.compressionQueue.push(point);
|
||||
|
||||
// 当压缩队列达到压缩比例时，执行压缩
|
||||
if (this.compressionQueue.length >= this.config.compressionRatio) {
|
||||
this.compressData();
|
||||
}
|
||||
|
||||
// 清理过期的压缩数据
|
||||
const compressedCutoff =
|
||||
Date.now() - this.config.compressedDataMinutes * 60 * 1000;
|
||||
this.compressedBuffer = this.compressedBuffer.filter(
|
||||
(p) => p.timestamp > compressedCutoff,
|
||||
);
|
||||
}
|
||||
|
||||
private compressData(): void {
|
||||
if (this.compressionQueue.length === 0) return;
|
||||
|
||||
// 计算平均值进行压缩
|
||||
const totalUp = this.compressionQueue.reduce((sum, p) => sum + p.up, 0);
|
||||
const totalDown = this.compressionQueue.reduce((sum, p) => sum + p.down, 0);
|
||||
const avgTimestamp =
|
||||
this.compressionQueue.reduce((sum, p) => sum + p.timestamp, 0) /
|
||||
this.compressionQueue.length;
|
||||
|
||||
const compressedPoint: ICompressedDataPoint = {
|
||||
up: totalUp / this.compressionQueue.length,
|
||||
down: totalDown / this.compressionQueue.length,
|
||||
timestamp: avgTimestamp,
|
||||
samples: this.compressionQueue.length,
|
||||
};
|
||||
|
||||
this.compressedBuffer.push(compressedPoint);
|
||||
this.compressionQueue = [];
|
||||
|
||||
debugLog(`[DataSampler] 压缩了 ${compressedPoint.samples} 个数据点`);
|
||||
}
|
||||
|
||||
getDataForTimeRange(minutes: number): ITrafficDataPoint[] {
|
||||
const cutoff = Date.now() - minutes * 60 * 1000;
|
||||
|
||||
// 如果请求的时间范围在原始数据范围内，直接返回原始数据
|
||||
if (minutes <= this.config.rawDataMinutes) {
|
||||
return this.rawBuffer.filter((p) => p.timestamp > cutoff);
|
||||
}
|
||||
|
||||
// 否则组合原始数据和压缩数据
|
||||
const rawData = this.rawBuffer.filter((p) => p.timestamp > cutoff);
|
||||
const compressedData = this.compressedBuffer
|
||||
.filter(
|
||||
(p) =>
|
||||
p.timestamp > cutoff &&
|
||||
p.timestamp <= Date.now() - this.config.rawDataMinutes * 60 * 1000,
|
||||
)
|
||||
.map((p) => ({
|
||||
up: p.up,
|
||||
down: p.down,
|
||||
timestamp: p.timestamp,
|
||||
name: new Date(p.timestamp).toLocaleTimeString("en-US", {
|
||||
hour12: false,
|
||||
hour: "2-digit",
|
||||
minute: "2-digit",
|
||||
second: "2-digit",
|
||||
}),
|
||||
}));
|
||||
|
||||
return [...compressedData, ...rawData].sort(
|
||||
(a, b) => a.timestamp - b.timestamp,
|
||||
);
|
||||
}
|
||||
|
||||
getStats() {
|
||||
return {
|
||||
rawBufferSize: this.rawBuffer.length,
|
||||
compressedBufferSize: this.compressedBuffer.length,
|
||||
compressionQueueSize: this.compressionQueue.length,
|
||||
totalMemoryPoints: this.rawBuffer.length + this.compressedBuffer.length,
|
||||
};
|
||||
}
|
||||
|
||||
clear(): void {
|
||||
this.rawBuffer = [];
|
||||
this.compressedBuffer = [];
|
||||
this.compressionQueue = [];
|
||||
}
|
||||
}
|
||||
|
||||
// 全局单例
|
||||
const refCounter = new ReferenceCounter();
|
||||
let globalSampler: TrafficDataSampler | null = null;
|
||||
|
||||
/**
|
||||
* 增强的流量监控Hook - 支持数据压缩、采样和引用计数
|
||||
*/
|
||||
export const useTrafficMonitorEnhanced = () => {
|
||||
// 初始化采样器
|
||||
if (!globalSampler) {
|
||||
globalSampler = new TrafficDataSampler({
|
||||
rawDataMinutes: 10, // 原始数据保持10分钟
|
||||
compressedDataMinutes: 60, // 压缩数据保持1小时
|
||||
compressionRatio: 5, // 每5个原始点压缩成1个
|
||||
start(rangeMinutes?: number) {
|
||||
this.currentRange = rangeMinutes ?? this.currentRange;
|
||||
this.handle({
|
||||
type: "init",
|
||||
config: {
|
||||
...this.config,
|
||||
defaultRangeMinutes: this.currentRange,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
const [, forceRender] = useReducer((version: number) => version + 1, 0);
|
||||
const cleanupRef = useRef<(() => void) | null>(null);
|
||||
stop() {
|
||||
if (this.throttleTimer !== null) {
|
||||
clearTimeout(this.throttleTimer);
|
||||
this.throttleTimer = null;
|
||||
}
|
||||
this.sampler.clear();
|
||||
this.lastTimestamp = undefined;
|
||||
}
|
||||
|
||||
const bumpRenderVersion = useCallback(() => {
|
||||
forceRender();
|
||||
}, []);
|
||||
handle(message: TrafficWorkerRequestMessage) {
|
||||
switch (message.type) {
|
||||
case "init": {
|
||||
this.config = { ...message.config };
|
||||
this.sampler = new TrafficDataSampler(this.config);
|
||||
this.currentRange = message.config.defaultRangeMinutes;
|
||||
this.emitSnapshot("init");
|
||||
break;
|
||||
}
|
||||
case "append": {
|
||||
const timestamp = message.payload.timestamp ?? Date.now();
|
||||
const dataPoint: ITrafficDataPoint = {
|
||||
up: message.payload.up || 0,
|
||||
down: message.payload.down || 0,
|
||||
timestamp,
|
||||
name: formatTrafficName(timestamp),
|
||||
};
|
||||
|
||||
// 注册引用计数
|
||||
this.lastTimestamp = timestamp;
|
||||
this.sampler.addDataPoint(dataPoint);
|
||||
this.scheduleSnapshot("append-throttle");
|
||||
break;
|
||||
}
|
||||
case "clear": {
|
||||
this.sampler.clear();
|
||||
this.lastTimestamp = undefined;
|
||||
this.emitSnapshot("clear");
|
||||
break;
|
||||
}
|
||||
case "setRange": {
|
||||
if (this.currentRange !== message.minutes) {
|
||||
this.currentRange = message.minutes;
|
||||
this.emitSnapshot("range-change");
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "requestSnapshot": {
|
||||
this.emitSnapshot("request");
|
||||
break;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private emitSnapshot(reason: ITrafficWorkerSnapshotMessage["reason"]) {
|
||||
const dataPoints = this.sampler.getDataForTimeRange(this.currentRange);
|
||||
const availableDataPoints = this.sampler.getDataForTimeRange(
|
||||
this.config.compressedDataMinutes,
|
||||
);
|
||||
|
||||
this.emit({
|
||||
type: "snapshot",
|
||||
dataPoints,
|
||||
availableDataPoints,
|
||||
samplerStats: this.sampler.getStats(),
|
||||
rangeMinutes: this.currentRange,
|
||||
lastTimestamp: this.lastTimestamp,
|
||||
reason,
|
||||
});
|
||||
}
|
||||
|
||||
private scheduleSnapshot(reason: ITrafficWorkerSnapshotMessage["reason"]) {
|
||||
if (this.throttleTimer !== null) return;
|
||||
this.throttleTimer = setTimeout(() => {
|
||||
this.throttleTimer = null;
|
||||
this.emitSnapshot(reason);
|
||||
}, this.config.snapshotIntervalMs);
|
||||
}
|
||||
}
|
||||
|
||||
class TrafficWorkerClient {
|
||||
private worker: Worker | null = null;
|
||||
private inlineMonitor: InlineTrafficMonitor | null = null;
|
||||
private mode: "worker" | "inline" | null = null;
|
||||
private listeners = new Set<
|
||||
(snapshot: ITrafficWorkerSnapshotMessage) => void
|
||||
>();
|
||||
private pendingMessages: TrafficWorkerRequestMessage[] = [];
|
||||
private ready = false;
|
||||
private currentRange = WORKER_CONFIG.defaultRangeMinutes;
|
||||
|
||||
start(rangeMinutes?: number) {
|
||||
if (typeof window === "undefined") {
|
||||
debugLog("[TrafficWorkerClient] Window not available, skip start");
|
||||
return;
|
||||
}
|
||||
|
||||
this.currentRange = rangeMinutes ?? this.currentRange;
|
||||
|
||||
if (this.ready) return;
|
||||
|
||||
const initMessage: TrafficWorkerRequestMessage = {
|
||||
type: "init",
|
||||
config: {
|
||||
rawDataMinutes: WORKER_CONFIG.rawDataMinutes,
|
||||
compressedDataMinutes: WORKER_CONFIG.compressedDataMinutes,
|
||||
compressionRatio: WORKER_CONFIG.compressionRatio,
|
||||
snapshotIntervalMs: WORKER_CONFIG.snapshotIntervalMs,
|
||||
defaultRangeMinutes: this.currentRange,
|
||||
},
|
||||
};
|
||||
|
||||
if (typeof Worker !== "undefined") {
|
||||
try {
|
||||
this.worker = new Worker(
|
||||
new URL("../services/traffic-monitor-worker.ts", import.meta.url),
|
||||
{ type: "module" },
|
||||
);
|
||||
this.mode = "worker";
|
||||
|
||||
this.worker.onmessage = (
|
||||
event: MessageEvent<TrafficWorkerResponseMessage>,
|
||||
) => {
|
||||
const message = event.data;
|
||||
if (message.type === "snapshot") {
|
||||
this.listeners.forEach((listener) => listener(message));
|
||||
}
|
||||
};
|
||||
|
||||
this.worker.onerror = (error) => {
|
||||
debugLog(`[TrafficWorkerClient] Worker error: ${String(error)}`);
|
||||
};
|
||||
|
||||
this.ready = true;
|
||||
this.post(initMessage);
|
||||
this.flushQueue();
|
||||
return;
|
||||
} catch (error) {
|
||||
debugLog(
|
||||
`[TrafficWorkerClient] Worker initialization failed, falling back to inline sampler: ${String(error)}`,
|
||||
);
|
||||
this.worker = null;
|
||||
this.mode = null;
|
||||
}
|
||||
} else {
|
||||
debugLog(
|
||||
"[TrafficWorkerClient] Worker not supported, using inline sampler",
|
||||
);
|
||||
}
|
||||
|
||||
this.startInline(initMessage);
|
||||
}
|
||||
|
||||
private startInline(initMessage: TrafficWorkerRequestMessage) {
|
||||
this.inlineMonitor = new InlineTrafficMonitor((snapshot) =>
|
||||
this.listeners.forEach((listener) => listener(snapshot)),
|
||||
);
|
||||
this.mode = "inline";
|
||||
this.ready = true;
|
||||
this.post(initMessage);
|
||||
this.flushQueue();
|
||||
}
|
||||
|
||||
stop() {
|
||||
if (this.worker) {
|
||||
this.worker.terminate();
|
||||
}
|
||||
if (this.inlineMonitor) {
|
||||
this.inlineMonitor.stop();
|
||||
}
|
||||
this.worker = null;
|
||||
this.inlineMonitor = null;
|
||||
this.mode = null;
|
||||
this.ready = false;
|
||||
this.pendingMessages = [];
|
||||
}
|
||||
|
||||
onSnapshot(listener: (snapshot: ITrafficWorkerSnapshotMessage) => void) {
|
||||
this.listeners.add(listener);
|
||||
return () => {
|
||||
this.listeners.delete(listener);
|
||||
};
|
||||
}
|
||||
|
||||
private post(message: TrafficWorkerRequestMessage) {
|
||||
if (!this.ready) {
|
||||
this.pendingMessages.push(message);
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.mode === "worker" && this.worker) {
|
||||
this.worker.postMessage(message);
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.mode === "inline" && this.inlineMonitor) {
|
||||
this.inlineMonitor.handle(message);
|
||||
return;
|
||||
}
|
||||
|
||||
this.pendingMessages.push(message);
|
||||
}
|
||||
|
||||
private flushQueue() {
|
||||
if (!this.ready || this.pendingMessages.length === 0) {
|
||||
return;
|
||||
}
|
||||
const queued = [...this.pendingMessages];
|
||||
this.pendingMessages = [];
|
||||
queued.forEach((message) => {
|
||||
this.post(message);
|
||||
});
|
||||
}
|
||||
|
||||
private ensureStarted() {
|
||||
if (!this.ready) {
|
||||
this.start(this.currentRange);
|
||||
}
|
||||
}
|
||||
|
||||
appendData(traffic: Traffic) {
|
||||
this.ensureStarted();
|
||||
this.post({
|
||||
type: "append",
|
||||
payload: {
|
||||
up: traffic?.up ?? 0,
|
||||
down: traffic?.down ?? 0,
|
||||
timestamp: Date.now(),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
clearData() {
|
||||
this.ensureStarted();
|
||||
this.post({ type: "clear" });
|
||||
}
|
||||
|
||||
setRange(minutes: number) {
|
||||
this.currentRange = minutes;
|
||||
this.post({ type: "setRange", minutes });
|
||||
}
|
||||
|
||||
requestSnapshot() {
|
||||
this.ensureStarted();
|
||||
this.post({ type: "requestSnapshot" });
|
||||
}
|
||||
}
|
||||
|
||||
const refCounter = new ReferenceCounter();
|
||||
let workerClient: TrafficWorkerClient | null = null;
|
||||
const getWorkerClient = () => {
|
||||
if (!workerClient) {
|
||||
workerClient = new TrafficWorkerClient();
|
||||
}
|
||||
return workerClient;
|
||||
};
|
||||
|
||||
const EMPTY_STATS: ISamplerStats = {
|
||||
rawBufferSize: 0,
|
||||
compressedBufferSize: 0,
|
||||
compressionQueueSize: 0,
|
||||
totalMemoryPoints: 0,
|
||||
};
|
||||
|
||||
/**
|
||||
* 增强的流量监控Hook - Web Worker驱动的数据采样与压缩
|
||||
*/
|
||||
export const useTrafficMonitorEnhanced = (options?: {
|
||||
subscribe?: boolean;
|
||||
}) => {
|
||||
const subscribeToSnapshots = options?.subscribe ?? true;
|
||||
const [latestSnapshot, setLatestSnapshot] = useState<{
|
||||
availableDataPoints: ITrafficDataPoint[];
|
||||
samplerStats: ISamplerStats;
|
||||
lastTimestamp?: number;
|
||||
}>({
|
||||
availableDataPoints: [],
|
||||
samplerStats: EMPTY_STATS,
|
||||
lastTimestamp: undefined,
|
||||
});
|
||||
const [rangeMinutes, setRangeMinutes] = useState(
|
||||
WORKER_CONFIG.defaultRangeMinutes,
|
||||
);
|
||||
const [now, setNow] = useState(() => Date.now());
|
||||
const [, forceRefCountRender] = useReducer((value) => value + 1, 0);
|
||||
|
||||
const clientRef = useRef<TrafficWorkerClient | null>(getWorkerClient());
|
||||
const currentRangeRef = useRef<number>(WORKER_CONFIG.defaultRangeMinutes);
|
||||
|
||||
// 注册引用计数与Worker生命周期
|
||||
useEffect(() => {
|
||||
debugLog("[TrafficMonitorEnhanced] 组件挂载，注册引用计数");
|
||||
const client = getWorkerClient();
|
||||
clientRef.current = client;
|
||||
|
||||
const stopWatchRefCount = refCounter.onCountChange(() =>
|
||||
forceRefCountRender(),
|
||||
);
|
||||
const cleanup = refCounter.increment();
|
||||
cleanupRef.current = cleanup;
|
||||
client.start(currentRangeRef.current);
|
||||
|
||||
let unsubscribe: (() => void) | undefined;
|
||||
if (subscribeToSnapshots) {
|
||||
unsubscribe = client.onSnapshot((message) => {
|
||||
setLatestSnapshot({
|
||||
availableDataPoints:
|
||||
message.availableDataPoints ?? message.dataPoints,
|
||||
samplerStats: message.samplerStats,
|
||||
lastTimestamp: message.lastTimestamp,
|
||||
});
|
||||
});
|
||||
|
||||
client.requestSnapshot();
|
||||
}
|
||||
|
||||
return () => {
|
||||
debugLog("[TrafficMonitorEnhanced] 组件卸载，清理引用计数");
|
||||
unsubscribe?.();
|
||||
stopWatchRefCount();
|
||||
cleanup();
|
||||
cleanupRef.current = null;
|
||||
};
|
||||
}, []);
|
||||
|
||||
// 设置引用计数变化回调
|
||||
useEffect(() => {
|
||||
const handleCountChange = () => {
|
||||
debugLog(
|
||||
`[TrafficMonitorEnhanced] 引用计数变化: ${refCounter.getCount()}`,
|
||||
);
|
||||
if (refCounter.getCount() === 0) {
|
||||
debugLog("[TrafficMonitorEnhanced] 所有组件已卸载，暂停数据收集");
|
||||
} else {
|
||||
debugLog("[TrafficMonitorEnhanced] 开始数据收集");
|
||||
client.stop();
|
||||
}
|
||||
};
|
||||
}, [subscribeToSnapshots]);
|
||||
|
||||
refCounter.onCountChange(handleCountChange);
|
||||
}, []);
|
||||
// Periodically refresh "now" so idle streams age out of the selected window when subscribed
|
||||
useEffect(() => {
|
||||
if (!subscribeToSnapshots) return;
|
||||
|
||||
// 获取指定时间范围的数据
|
||||
const getDataForTimeRange = useCallback(
|
||||
(minutes: number): ITrafficDataPoint[] => {
|
||||
if (!globalSampler) return [];
|
||||
return globalSampler.getDataForTimeRange(minutes);
|
||||
},
|
||||
[],
|
||||
);
|
||||
const timer = window.setInterval(() => {
|
||||
setNow(Date.now());
|
||||
}, 1000);
|
||||
|
||||
return () => window.clearInterval(timer);
|
||||
}, [subscribeToSnapshots]);
|
||||
|
||||
// 添加流量数据
|
||||
const appendData = useCallback((traffic: Traffic) => {
|
||||
if (globalSampler) {
|
||||
// 添加到采样器
|
||||
const timestamp = Date.now();
|
||||
const dataPoint: ITrafficDataPoint = {
|
||||
up: traffic?.up || 0,
|
||||
down: traffic?.down || 0,
|
||||
timestamp,
|
||||
name: new Date(timestamp).toLocaleTimeString("en-US", {
|
||||
hour12: false,
|
||||
hour: "2-digit",
|
||||
minute: "2-digit",
|
||||
second: "2-digit",
|
||||
}),
|
||||
};
|
||||
globalSampler.addDataPoint(dataPoint);
|
||||
}
|
||||
clientRef.current?.appendData(traffic);
|
||||
}, []);
|
||||
|
||||
// 请求不同时间范围的数据
|
||||
const requestRange = useCallback((minutes: number) => {
|
||||
currentRangeRef.current = minutes;
|
||||
setRangeMinutes(minutes);
|
||||
clientRef.current?.setRange(minutes);
|
||||
}, []);
|
||||
|
||||
// 清空数据
|
||||
const clearData = useCallback(() => {
|
||||
if (globalSampler) {
|
||||
globalSampler.clear();
|
||||
bumpRenderVersion();
|
||||
}
|
||||
}, [bumpRenderVersion]);
|
||||
|
||||
// 获取采样器统计信息
|
||||
const getSamplerStats = useCallback(() => {
|
||||
return (
|
||||
globalSampler?.getStats() || {
|
||||
rawBufferSize: 0,
|
||||
compressedBufferSize: 0,
|
||||
compressionQueueSize: 0,
|
||||
totalMemoryPoints: 0,
|
||||
}
|
||||
);
|
||||
clientRef.current?.clearData();
|
||||
}, []);
|
||||
|
||||
const filteredDataPoints = useMemo(() => {
|
||||
const sourceData = latestSnapshot.availableDataPoints;
|
||||
if (sourceData.length === 0) return [];
|
||||
|
||||
const cutoff = now - rangeMinutes * 60 * 1000;
|
||||
return sourceData.filter((point) => point.timestamp > cutoff);
|
||||
}, [latestSnapshot.availableDataPoints, rangeMinutes, now]);
|
||||
|
||||
return {
|
||||
// 图表数据管理
|
||||
graphData: {
|
||||
dataPoints: globalSampler?.getDataForTimeRange(60) || [], // 默认获取1小时数据
|
||||
getDataForTimeRange,
|
||||
dataPoints: filteredDataPoints,
|
||||
currentRangeMinutes: rangeMinutes,
|
||||
requestRange,
|
||||
appendData,
|
||||
clearData,
|
||||
},
|
||||
// 性能统计
|
||||
samplerStats: getSamplerStats(),
|
||||
samplerStats: latestSnapshot.samplerStats,
|
||||
referenceCount: refCounter.getCount(),
|
||||
};
|
||||
};
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
import type {
|
||||
ISamplingConfig,
|
||||
ITrafficDataPoint,
|
||||
ITrafficWorkerSnapshotMessage,
|
||||
TrafficWorkerRequestMessage,
|
||||
} from "../types/traffic-monitor";
|
||||
import {
|
||||
TrafficDataSampler,
|
||||
formatTrafficName,
|
||||
} from "../utils/traffic-sampler";
|
||||
|
||||
const DEFAULT_CONFIG: ISamplingConfig & {
|
||||
snapshotIntervalMs: number;
|
||||
defaultRangeMinutes: number;
|
||||
} = {
|
||||
rawDataMinutes: 10,
|
||||
compressedDataMinutes: 60,
|
||||
compressionRatio: 5,
|
||||
snapshotIntervalMs: 250,
|
||||
defaultRangeMinutes: 10,
|
||||
};
|
||||
|
||||
interface WorkerScope {
|
||||
postMessage: (message: unknown) => void;
|
||||
onmessage:
|
||||
| ((event: MessageEvent<TrafficWorkerRequestMessage>) => void)
|
||||
| null;
|
||||
setTimeout: typeof setTimeout;
|
||||
clearTimeout: typeof clearTimeout;
|
||||
}
|
||||
|
||||
const ctx: WorkerScope = self as unknown as WorkerScope;
|
||||
|
||||
let config = { ...DEFAULT_CONFIG };
|
||||
let sampler = new TrafficDataSampler(config);
|
||||
let currentRangeMinutes = config.defaultRangeMinutes;
|
||||
let throttleTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
let lastTimestamp: number | undefined;
|
||||
|
||||
const broadcastSnapshot = (reason: ITrafficWorkerSnapshotMessage["reason"]) => {
|
||||
const dataPoints = sampler.getDataForTimeRange(currentRangeMinutes);
|
||||
const availableDataPoints = sampler.getDataForTimeRange(
|
||||
config.compressedDataMinutes,
|
||||
);
|
||||
const samplerStats = sampler.getStats();
|
||||
|
||||
const message: ITrafficWorkerSnapshotMessage = {
|
||||
type: "snapshot",
|
||||
dataPoints,
|
||||
availableDataPoints,
|
||||
samplerStats,
|
||||
rangeMinutes: currentRangeMinutes,
|
||||
lastTimestamp,
|
||||
reason,
|
||||
};
|
||||
|
||||
ctx.postMessage(message);
|
||||
};
|
||||
|
||||
const scheduleSnapshot = (reason: ITrafficWorkerSnapshotMessage["reason"]) => {
|
||||
if (throttleTimer !== null) return;
|
||||
throttleTimer = ctx.setTimeout(() => {
|
||||
throttleTimer = null;
|
||||
broadcastSnapshot(reason);
|
||||
}, config.snapshotIntervalMs);
|
||||
};
|
||||
|
||||
ctx.onmessage = (event: MessageEvent<TrafficWorkerRequestMessage>) => {
|
||||
const message = event.data;
|
||||
|
||||
switch (message.type) {
|
||||
case "init": {
|
||||
config = { ...message.config };
|
||||
sampler = new TrafficDataSampler(config);
|
||||
currentRangeMinutes = message.config.defaultRangeMinutes;
|
||||
broadcastSnapshot("init");
|
||||
break;
|
||||
}
|
||||
case "append": {
|
||||
const timestamp = message.payload.timestamp ?? Date.now();
|
||||
const dataPoint: ITrafficDataPoint = {
|
||||
up: message.payload.up || 0,
|
||||
down: message.payload.down || 0,
|
||||
timestamp,
|
||||
name: formatTrafficName(timestamp),
|
||||
};
|
||||
|
||||
lastTimestamp = timestamp;
|
||||
sampler.addDataPoint(dataPoint);
|
||||
scheduleSnapshot("append-throttle");
|
||||
break;
|
||||
}
|
||||
case "clear": {
|
||||
sampler.clear();
|
||||
lastTimestamp = undefined;
|
||||
broadcastSnapshot("clear");
|
||||
break;
|
||||
}
|
||||
case "setRange": {
|
||||
if (currentRangeMinutes !== message.minutes) {
|
||||
currentRangeMinutes = message.minutes;
|
||||
broadcastSnapshot("range-change");
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "requestSnapshot": {
|
||||
broadcastSnapshot("request");
|
||||
break;
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
};
|
||||
@@ -0,0 +1,81 @@
|
||||
export interface ITrafficDataPoint {
|
||||
up: number;
|
||||
down: number;
|
||||
timestamp: number;
|
||||
name: string;
|
||||
}
|
||||
|
||||
export interface ISamplingConfig {
|
||||
rawDataMinutes: number;
|
||||
compressedDataMinutes: number;
|
||||
compressionRatio: number;
|
||||
}
|
||||
|
||||
export interface ISamplerStats {
|
||||
rawBufferSize: number;
|
||||
compressedBufferSize: number;
|
||||
compressionQueueSize: number;
|
||||
totalMemoryPoints: number;
|
||||
}
|
||||
|
||||
export interface ITrafficWorkerInitMessage {
|
||||
type: "init";
|
||||
config: ISamplingConfig & {
|
||||
snapshotIntervalMs: number;
|
||||
defaultRangeMinutes: number;
|
||||
};
|
||||
}
|
||||
|
||||
export interface ITrafficWorkerAppendMessage {
|
||||
type: "append";
|
||||
payload: {
|
||||
up: number;
|
||||
down: number;
|
||||
timestamp?: number;
|
||||
};
|
||||
}
|
||||
|
||||
export interface ITrafficWorkerClearMessage {
|
||||
type: "clear";
|
||||
}
|
||||
|
||||
export interface ITrafficWorkerSetRangeMessage {
|
||||
type: "setRange";
|
||||
minutes: number;
|
||||
}
|
||||
|
||||
export interface ITrafficWorkerRequestSnapshotMessage {
|
||||
type: "requestSnapshot";
|
||||
}
|
||||
|
||||
export type TrafficWorkerRequestMessage =
|
||||
| ITrafficWorkerInitMessage
|
||||
| ITrafficWorkerAppendMessage
|
||||
| ITrafficWorkerClearMessage
|
||||
| ITrafficWorkerSetRangeMessage
|
||||
| ITrafficWorkerRequestSnapshotMessage;
|
||||
|
||||
export interface ITrafficWorkerSnapshotMessage {
|
||||
type: "snapshot";
|
||||
dataPoints: ITrafficDataPoint[];
|
||||
availableDataPoints: ITrafficDataPoint[];
|
||||
samplerStats: ISamplerStats;
|
||||
rangeMinutes: number;
|
||||
lastTimestamp?: number;
|
||||
reason:
|
||||
| "init"
|
||||
| "interval"
|
||||
| "range-change"
|
||||
| "request"
|
||||
| "append-throttle"
|
||||
| "clear";
|
||||
}
|
||||
|
||||
export interface ITrafficWorkerLogMessage {
|
||||
type: "log";
|
||||
message: string;
|
||||
}
|
||||
|
||||
export type TrafficWorkerResponseMessage =
|
||||
| ITrafficWorkerSnapshotMessage
|
||||
| ITrafficWorkerLogMessage;
|
||||
@@ -0,0 +1,107 @@
|
||||
import type {
|
||||
ISamplerStats,
|
||||
ISamplingConfig,
|
||||
ITrafficDataPoint,
|
||||
} from "@/types/traffic-monitor";
|
||||
|
||||
interface ICompressedDataPoint {
|
||||
up: number;
|
||||
down: number;
|
||||
timestamp: number;
|
||||
samples: number;
|
||||
}
|
||||
|
||||
export const formatTrafficName = (timestamp: number) =>
|
||||
new Date(timestamp).toLocaleTimeString("en-US", {
|
||||
hour12: false,
|
||||
hour: "2-digit",
|
||||
minute: "2-digit",
|
||||
second: "2-digit",
|
||||
});
|
||||
|
||||
export class TrafficDataSampler {
|
||||
private rawBuffer: ITrafficDataPoint[] = [];
|
||||
private compressedBuffer: ICompressedDataPoint[] = [];
|
||||
private compressionQueue: ITrafficDataPoint[] = [];
|
||||
|
||||
constructor(private config: ISamplingConfig) {}
|
||||
|
||||
addDataPoint(point: ITrafficDataPoint) {
|
||||
this.rawBuffer.push(point);
|
||||
|
||||
const rawCutoff = Date.now() - this.config.rawDataMinutes * 60 * 1000;
|
||||
this.rawBuffer = this.rawBuffer.filter((p) => p.timestamp > rawCutoff);
|
||||
|
||||
this.compressionQueue.push(point);
|
||||
if (this.compressionQueue.length >= this.config.compressionRatio) {
|
||||
this.compressData();
|
||||
}
|
||||
|
||||
const compressedCutoff =
|
||||
Date.now() - this.config.compressedDataMinutes * 60 * 1000;
|
||||
this.compressedBuffer = this.compressedBuffer.filter(
|
||||
(p) => p.timestamp > compressedCutoff,
|
||||
);
|
||||
}
|
||||
|
||||
private compressData() {
|
||||
if (this.compressionQueue.length === 0) return;
|
||||
|
||||
const totalUp = this.compressionQueue.reduce((sum, p) => sum + p.up, 0);
|
||||
const totalDown = this.compressionQueue.reduce((sum, p) => sum + p.down, 0);
|
||||
const avgTimestamp =
|
||||
this.compressionQueue.reduce((sum, p) => sum + p.timestamp, 0) /
|
||||
this.compressionQueue.length;
|
||||
|
||||
const compressedPoint: ICompressedDataPoint = {
|
||||
up: totalUp / this.compressionQueue.length,
|
||||
down: totalDown / this.compressionQueue.length,
|
||||
timestamp: avgTimestamp,
|
||||
samples: this.compressionQueue.length,
|
||||
};
|
||||
|
||||
this.compressedBuffer.push(compressedPoint);
|
||||
this.compressionQueue = [];
|
||||
}
|
||||
|
||||
getDataForTimeRange(minutes: number): ITrafficDataPoint[] {
|
||||
const cutoff = Date.now() - minutes * 60 * 1000;
|
||||
const rawData = this.rawBuffer.filter((p) => p.timestamp > cutoff);
|
||||
|
||||
if (minutes <= this.config.rawDataMinutes) {
|
||||
return rawData;
|
||||
}
|
||||
|
||||
const compressedData = this.compressedBuffer
|
||||
.filter(
|
||||
(p) =>
|
||||
p.timestamp > cutoff &&
|
||||
p.timestamp <= Date.now() - this.config.rawDataMinutes * 60 * 1000,
|
||||
)
|
||||
.map((p) => ({
|
||||
up: p.up,
|
||||
down: p.down,
|
||||
timestamp: p.timestamp,
|
||||
name: formatTrafficName(p.timestamp),
|
||||
}));
|
||||
|
||||
return [...compressedData, ...rawData].sort(
|
||||
(a, b) => a.timestamp - b.timestamp,
|
||||
);
|
||||
}
|
||||
|
||||
getStats(): ISamplerStats {
|
||||
return {
|
||||
rawBufferSize: this.rawBuffer.length,
|
||||
compressedBufferSize: this.compressedBuffer.length,
|
||||
compressionQueueSize: this.compressionQueue.length,
|
||||
totalMemoryPoints: this.rawBuffer.length + this.compressedBuffer.length,
|
||||
};
|
||||
}
|
||||
|
||||
clear() {
|
||||
this.rawBuffer = [];
|
||||
this.compressedBuffer = [];
|
||||
this.compressionQueue = [];
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user