From 7ab0e426264707fb85ee38d0a9a73caf41899635 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 29 八月 2026 13:11:40 +0800
Subject: [PATCH] 实现采集器独立实时与历史采样模拟
---
ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java | 73 ++++++++++++++++++++++++++++++++++--
1 files changed, 69 insertions(+), 4 deletions(-)
diff --git a/ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java b/ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java
index c299502..ddac794 100644
--- a/ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java
+++ b/ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java
@@ -12,11 +12,18 @@
import com.ruoyi.device.simulator.BtsSimulatorStateStore;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelTombstone;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorChannelState;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryFaults;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMetricState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.DeviceState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.RecentRequest;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.SimulatorState;
import com.ruoyi.device.simulator.InMemoryBtsSimulatorStateStore;
+import com.ruoyi.device.simulator.DevelopmentCollectorSimulation;
+import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorHistoryPage;
+import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;
/** 提供由开发后端托管的临时只读 BTS 模拟端点。 */
public final class BtsDevelopmentSimulator implements AutoCloseable
@@ -138,8 +145,7 @@
CollectorState current = requireCollector(collectorId);
List<CollectorState> collectors = new ArrayList<>(state.collectors());
collectors.replaceAll(item -> item.collectorId() == collectorId
- ? new CollectorState(collectorId, current.deviceCode(), ip.trim(), current.mac(), current.name(),
- current.manufacturer(), current.model(), channelCount, online)
+ ? DevelopmentCollectorSimulation.resize(current, ip, channelCount, online, Instant.now())
: item);
persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
state.nextCollectorId(), state.tombstones()));
@@ -155,6 +161,45 @@
state.nextCollectorId(), state.tombstones()));
}
+ /** 配置采集器独立采样频率和历史下载故障场景。 */
+ public synchronized void configureCollectorSampling(int collectorId, int sampleIntervalMs,
+ CollectorHistoryFaults historyFaults)
+ {
+ CollectorState current = requireCollector(collectorId);
+ replaceCollector(DevelopmentCollectorSimulation.configureSampling(current, sampleIntervalMs, historyFaults));
+ }
+
+ /** 配置一个稳定 S 通道的固定或动态演示指标。 */
+ public synchronized void configureCollectorChannel(int collectorId, String channelCode,
+ List<CollectorMetricState> metrics)
+ {
+ CollectorState current = requireCollector(collectorId);
+ replaceCollector(DevelopmentCollectorSimulation.configureChannel(current, channelCode, metrics));
+ }
+
+ /** 按每台采集器自己的频率推进实时和历史采样。 */
+ public synchronized void advanceCollectors(Instant now)
+ {
+ List<CollectorState> collectors = state.collectors().stream()
+ .map(collector -> DevelopmentCollectorSimulation.advance(collector, now))
+ .toList();
+ if (!collectors.equals(state.collectors()))
+ persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
+ state.nextCollectorId(), state.tombstones()));
+ }
+
+ /** 返回采集器最新实时快照；快照与分页历史保持不同读取路径。 */
+ public synchronized CollectorRealtimeSnapshot collectorRealtime(int collectorId)
+ {
+ return DevelopmentCollectorSimulation.realtime(requireCollector(collectorId));
+ }
+
+ /** 分页下载采集器模拟历史，并应用已配置的重复、冲突、乱序或缺失场景。 */
+ public synchronized CollectorHistoryPage collectorHistory(int collectorId, long offset, int limit)
+ {
+ return DevelopmentCollectorSimulation.history(requireCollector(collectorId), offset, limit);
+ }
+
/** 返回控制页需要的运行状态快照。 */
public synchronized StateSnapshot snapshot()
{
@@ -168,7 +213,9 @@
List<CollectorSnapshot> collectorSnapshots = state.collectors().stream()
.map(collector -> new CollectorSnapshot(collector.collectorId(), collector.deviceCode(),
collector.ip(), collector.mac(), collector.name(), collector.manufacturer(),
- collector.model(), collector.channelCount(), collector.online()))
+ collector.model(), collector.channelCount(), collector.online(), collector.sampleIntervalMs(),
+ collector.channels(), collector.historyFaults(), collector.history().size(),
+ DevelopmentCollectorSimulation.realtime(collector).measurements()))
.toList();
return new StateSnapshot(running, HOST + ":" + port(), deviceSnapshots, collectorSnapshots,
stateStore.loadRecentRequests());
@@ -202,6 +249,14 @@
return new CollectorState(collectorId, "dev-sim-collector-" + collectorId,
ip.trim(), simulatorCollectorMac(collectorId), "dev-sim 采集器 " + collectorId,
"开发模拟器", "Virtual Collector", channelCount, true);
+ }
+
+ private void replaceCollector(CollectorState replacement)
+ {
+ List<CollectorState> collectors = new ArrayList<>(state.collectors());
+ collectors.replaceAll(item -> item.collectorId() == replacement.collectorId() ? replacement : item);
+ persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
+ state.nextCollectorId(), state.tombstones()));
}
private String respond(String request)
@@ -442,6 +497,7 @@
|| !isIpv4(collector.ip()) || !ips.add(collector.ip())
|| collector.channelCount() < 1 || collector.channelCount() > MAX_CHANNEL_COUNT)
throw new IllegalStateException("Invalid BTS simulator collector state in Redis");
+ DevelopmentCollectorSimulation.validate(collector);
maximumCollectorId = Math.max(maximumCollectorId, collector.collectorId());
}
java.util.Set<String> tombstoneIds = new java.util.HashSet<>();
@@ -471,6 +527,15 @@
public record ChannelSnapshot(int channelId, boolean online, DeviceWorkStatus workStatus) { }
public record CollectorSnapshot(int collectorId, String deviceCode, String ip, String mac, String name,
- String manufacturer, String model, int channelCount, boolean online) { }
+ String manufacturer, String model, int channelCount, boolean online, int sampleIntervalMs,
+ List<CollectorChannelState> channels, CollectorHistoryFaults historyFaults, int historyRecordCount,
+ List<CollectorMeasurementRecord> latestMeasurements)
+ {
+ public CollectorSnapshot
+ {
+ channels = List.copyOf(channels);
+ latestMeasurements = List.copyOf(latestMeasurements);
+ }
+ }
}
--
Gitblit v1.7.1