From 370ef4f5e663fdfc1d19d2774dfc4ec1ed0af60e Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 29 八月 2026 14:57:48 +0800
Subject: [PATCH] 修复采集器历史状态持续序列化错误
---
ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java | 45 +++++++++++++++++++++++++++++++++------------
1 files changed, 33 insertions(+), 12 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 1663a40..5548804 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
@@ -13,6 +13,7 @@
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.CollectorHistoryAppend;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryFaults;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMetricState;
@@ -23,6 +24,7 @@
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.CollectorAdvance;
import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorHistoryPage;
import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;
@@ -58,6 +60,7 @@
this.stateStore = stateStore;
state = stateStore.loadOrInitialize(defaultState());
validateRestoredState();
+ stateStore.completeValidatedRestore(state);
simulator = ScriptedBtsTcpSimulator.responding(port, this::respond);
}
@@ -155,11 +158,13 @@
/** 删除指定开发采集器，不影响检测柜 TCP 发现结果。 */
public synchronized void deleteCollector(int collectorId)
{
- requireCollector(collectorId);
+ CollectorState collector = requireCollector(collectorId);
List<CollectorState> collectors = new ArrayList<>(state.collectors());
collectors.removeIf(item -> item.collectorId() == collectorId);
- persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
- state.nextCollectorId(), state.tombstones()));
+ SimulatorState next = new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
+ state.nextCollectorId(), state.tombstones());
+ stateStore.saveAndDeleteCollectorHistory(next, collector.deviceCode());
+ this.state = next;
}
/** 配置采集器独立采样频率和历史下载故障场景。 */
@@ -181,12 +186,18 @@
/** 按每台采集器自己的频率推进实时和历史采样。 */
public synchronized void advanceCollectors(Instant now)
{
- List<CollectorState> collectors = state.collectors().stream()
- .map(collector -> DevelopmentCollectorSimulation.advance(collector, now))
- .toList();
- if (!collectors.equals(state.collectors()))
+ List<CollectorState> collectors = new ArrayList<>(state.collectors().size());
+ List<CollectorHistoryAppend> historyAppends = new ArrayList<>();
+ for (CollectorState collector : state.collectors())
+ {
+ CollectorAdvance advance = DevelopmentCollectorSimulation.advance(collector, now);
+ collectors.add(advance.state());
+ if (!advance.appendedRecords().isEmpty())
+ historyAppends.add(new CollectorHistoryAppend(collector.deviceCode(), advance.appendedRecords()));
+ }
+ if (!historyAppends.isEmpty())
persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
- state.nextCollectorId(), state.tombstones()));
+ state.nextCollectorId(), state.tombstones()), historyAppends);
}
/** 返回采集器最新实时快照；快照与分页历史保持不同读取路径。 */
@@ -198,7 +209,9 @@
/** 分页下载采集器模拟历史，并应用已配置的重复、冲突、乱序或缺失场景。 */
public synchronized CollectorHistoryPage collectorHistory(int collectorId, long offset, int limit)
{
- return DevelopmentCollectorSimulation.history(requireCollector(collectorId), offset, limit);
+ CollectorState collector = requireCollector(collectorId);
+ return DevelopmentCollectorSimulation.history(collector,
+ stateStore.loadCollectorHistory(collector.deviceCode()), offset, limit);
}
/** 返回控制页需要的运行状态快照。 */
@@ -215,7 +228,7 @@
.map(collector -> new CollectorSnapshot(collector.collectorId(), collector.deviceCode(),
collector.ip(), collector.mac(), collector.name(), collector.manufacturer(),
collector.model(), collector.channelCount(), collector.online(), collector.sampleIntervalMs(),
- collector.channels(), collector.historyFaults(), collector.history().size(),
+ collector.channels(), collector.historyFaults(), collector.historyRecordCount(),
DevelopmentCollectorSimulation.realtime(collector).measurements()))
.toList();
return new StateSnapshot(running, HOST + ":" + port(), deviceSnapshots, collectorSnapshots,
@@ -452,6 +465,13 @@
state = nextState;
}
+ private void persist(SimulatorState nextState, List<CollectorHistoryAppend> historyAppends)
+ {
+ // 主状态与原始历史批次必须一起成功，内存序号才能向前推进。
+ stateStore.save(nextState, historyAppends);
+ state = nextState;
+ }
+
private String simulatorMac(int devtype, int devid)
{
return String.format("02:44:%02X:%02X:%02X:%02X",
@@ -498,7 +518,8 @@
|| !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);
+ DevelopmentCollectorSimulation.validate(collector,
+ stateStore.loadCollectorHistory(collector.deviceCode()));
maximumCollectorId = Math.max(maximumCollectorId, collector.collectorId());
}
java.util.Set<String> tombstoneIds = new java.util.HashSet<>();
@@ -529,7 +550,7 @@
public record CollectorSnapshot(int collectorId, String deviceCode, String ip, String mac, String name,
String manufacturer, String model, int channelCount, boolean online, int sampleIntervalMs,
- List<CollectorChannelState> channels, CollectorHistoryFaults historyFaults, int historyRecordCount,
+ List<CollectorChannelState> channels, CollectorHistoryFaults historyFaults, long historyRecordCount,
List<CollectorRealtimeMeasurement> latestMeasurements)
{
public CollectorSnapshot
--
Gitblit v1.7.1