修复采集器历史状态持续序列化错误

yeg

2026-08-29 370ef4f5e663fdfc1d19d2774dfc4ec1ed0af60e

 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