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

yeg

2026-08-29 370ef4f5e663fdfc1d19d2774dfc4ec1ed0af60e

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

将只追加原始历史拆分到独立 Redis 列表，并兼容迁移旧格式状态。补充恢复校验、分批读取、原子删除和真实序列化测试，避免长期采样触发状态对象大小上限。

 ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java

@@ -4,7 +4,10 @@
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson2.JSON;
@@ -19,9 +22,13 @@
{
    static final String STATE_KEY = "dev:bts-simulator:state:v1";
    static final String RECENT_REQUESTS_KEY = "dev:bts-simulator:recent-requests:v1";
    static final String HISTORY_KEYS_KEY = "dev:bts-simulator:collector-history-keys:v1";
    static final String HISTORY_KEY_PREFIX = "dev:bts-simulator:collector-history:v1:";
    private static final long RECENT_REQUESTS_TTL_HOURS = 1;
    private static final int HISTORY_READ_BATCH_SIZE = 1000;

    private final RedisTemplate<Object, Object> redisTemplate;
    private List<CollectorHistoryAppend> pendingEmbeddedHistory = List.of();

    public RedisBtsSimulatorStateStore(RedisTemplate<Object, Object> redisTemplate)
    {
@@ -32,18 +39,106 @@
    public SimulatorState loadOrInitialize(SimulatorState defaultState)
    {
        Object stored = redisTemplate.opsForValue().get(STATE_KEY);
        if (stored != null) return parseState(stored);
        if (stored != null) return restoreState(stored);
        String initial = JSON.toJSONString(defaultState);
        if (Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(STATE_KEY, initial))) return defaultState;
        stored = redisTemplate.opsForValue().get(STATE_KEY);
        if (stored == null) throw new IllegalStateException("BTS simulator state initialization lost in Redis");
        return parseState(stored);
        return restoreState(stored);
    }

    private SimulatorState restoreState(Object stored)
    {
        ParsedSimulatorState parsed = parseState(stored);
        // 旧格式历史先只供恢复校验读取，损坏状态不得被迁移写回覆盖。
        pendingEmbeddedHistory = parsed.embeddedHistory();
        return parsed.state();
    }

    @Override
    public void completeValidatedRestore(SimulatorState state)
    {
        if (pendingEmbeddedHistory.isEmpty()) return;
        save(state, pendingEmbeddedHistory);
        pendingEmbeddedHistory = List.of();
    }

    @Override
    public void save(SimulatorState state)
    {
        redisTemplate.opsForValue().set(STATE_KEY, JSON.toJSONString(state));
        save(state, List.of());
    }

    @Override
    @SuppressWarnings({ "rawtypes", "unchecked" })
    public void save(SimulatorState state, List<CollectorHistoryAppend> historyAppends)
    {
        List<Object> results = redisTemplate.execute(new SessionCallback<List<Object>>()
        {
            @Override
            public <K, V> List<Object> execute(RedisOperations<K, V> operations)
            {
                RedisOperations<Object, Object> typed = (RedisOperations) operations;
                typed.multi();
                for (CollectorHistoryAppend append : historyAppends)
                {
                    String historyKey = historyKey(append.collectorCode());
                    if (!append.records().isEmpty())
                    {
                        List<Object> records = new ArrayList<>(
                                append.records().stream().map(JSON::toJSONString).toList());
                        typed.opsForList().rightPushAll(historyKey, records);
                        typed.opsForSet().add(HISTORY_KEYS_KEY, historyKey);
                    }
                }
                typed.opsForValue().set(STATE_KEY, JSON.toJSONString(state));
                return typed.exec();
            }
        });
        if (results == null) throw new IllegalStateException("BTS simulator state transaction failed in Redis");
    }

    @Override
    public List<CollectorMeasurementRecord> loadCollectorHistory(String collectorCode)
    {
        String key = historyKey(collectorCode);
        Long storedSize = redisTemplate.opsForList().size(key);
        List<CollectorMeasurementRecord> records = new ArrayList<>();
        if (storedSize != null)
        {
            for (long start = 0; start < storedSize; start += HISTORY_READ_BATCH_SIZE)
            {
                long end = Math.min(start + HISTORY_READ_BATCH_SIZE, storedSize) - 1;
                List<Object> stored = redisTemplate.opsForList().range(key, start, end);
                if (stored != null)
                    stored.forEach(item -> records.add(collectorMeasurementRecord(item)));
            }
        }
        pendingEmbeddedHistory.stream()
                .filter(append -> append.collectorCode().equals(collectorCode))
                .forEach(append -> records.addAll(append.records()));
        return List.copyOf(records);
    }

    @Override
    @SuppressWarnings({ "rawtypes", "unchecked" })
    public void saveAndDeleteCollectorHistory(SimulatorState state, String collectorCode)
    {
        String historyKey = historyKey(collectorCode);
        List<Object> results = redisTemplate.execute(new SessionCallback<List<Object>>()
        {
            @Override
            public <K, V> List<Object> execute(RedisOperations<K, V> operations)
            {
                RedisOperations<Object, Object> typed = (RedisOperations) operations;
                typed.multi();
                typed.delete(historyKey);
                typed.opsForSet().remove(HISTORY_KEYS_KEY, historyKey);
                typed.opsForValue().set(STATE_KEY, JSON.toJSONString(state));
                return typed.exec();
            }
        });
        if (results == null) throw new IllegalStateException("BTS simulator collector deletion failed in Redis");
    }

    @Override
@@ -64,11 +159,15 @@
    @Override
    public void reset(SimulatorState defaultState)
    {
        pendingEmbeddedHistory = List.of();
        Set<Object> historyKeys = redisTemplate.opsForSet().members(HISTORY_KEYS_KEY);
        if (historyKeys != null && !historyKeys.isEmpty()) redisTemplate.delete(historyKeys);
        redisTemplate.delete(HISTORY_KEYS_KEY);
        redisTemplate.delete(RECENT_REQUESTS_KEY);
        save(defaultState);
    }

    private SimulatorState parseState(Object stored)
    private ParsedSimulatorState parseState(Object stored)
    {
        try
        {
@@ -80,11 +179,20 @@
            List<DeviceState> devices = new ArrayList<>();
            requiredArray(root, "devices").forEach(item -> devices.add(deviceState(item)));
            List<CollectorState> collectors = new ArrayList<>();
            requiredArray(root, "collectors").forEach(item -> collectors.add(collectorState(item)));
            List<CollectorHistoryAppend> embeddedHistory = new ArrayList<>();
            requiredArray(root, "collectors").forEach(item -> {
                JSONObject collectorObject = jsonObject(item);
                List<CollectorMeasurementRecord> history = embeddedCollectorHistory(collectorObject);
                CollectorState collector = collectorState(collectorObject, history);
                collectors.add(collector);
                if (!history.isEmpty())
                    embeddedHistory.add(new CollectorHistoryAppend(collector.deviceCode(), history));
            });
            List<ChannelTombstone> tombstones = new ArrayList<>();
            requiredArray(root, "tombstones").forEach(item -> tombstones.add(channelTombstone(item)));
            return new SimulatorState(version, devices, collectors,
            SimulatorState state = new SimulatorState(version, devices, collectors,
                    requiredInteger(root, "nextDeviceId"), requiredInteger(root, "nextCollectorId"), tombstones);
            return new ParsedSimulatorState(state, embeddedHistory);
        }
        catch (IllegalStateException exception)
        {
@@ -132,9 +240,17 @@
                object.getBooleanValue("online"), channels);
    }

    private CollectorState collectorState(Object value)
    private List<CollectorMeasurementRecord> embeddedCollectorHistory(JSONObject object)
    {
        JSONObject object = jsonObject(value);
        JSONArray values = object.getJSONArray("history");
        if (values == null) return List.of();
        List<CollectorMeasurementRecord> history = new ArrayList<>(values.size());
        values.forEach(item -> history.add(collectorMeasurementRecord(item)));
        return List.copyOf(history);
    }

    private CollectorState collectorState(JSONObject object, List<CollectorMeasurementRecord> embeddedHistory)
    {
        int collectorId = requiredInteger(object, "collectorId");
        String deviceCode = object.getString("deviceCode");
        String ip = object.getString("ip");
@@ -151,18 +267,19 @@

        List<CollectorChannelState> channels = new ArrayList<>();
        requiredArray(object, "channels").forEach(item -> channels.add(collectorChannelState(item)));
        List<CollectorMeasurementRecord> history = new ArrayList<>();
        requiredArray(object, "history").forEach(item -> history.add(collectorMeasurementRecord(item)));
        List<CollectorRealtimeMeasurement> latestMeasurements = new ArrayList<>();
        JSONArray latest = object.getJSONArray("latestMeasurements");
        if (latest != null)
            latest.forEach(item -> latestMeasurements.add(collectorRealtimeMeasurement(item)));
        else
            latestMeasurements.addAll(latestMeasurementsFromLegacyHistory(history, channels));
            latestMeasurements.addAll(latestMeasurementsFromLegacyHistory(embeddedHistory, channels));
        long historyRecordCount = object.containsKey("historyRecordCount")
                ? requiredLong(object, "historyRecordCount") : embeddedHistory.size();
        return new CollectorState(collectorId, deviceCode, ip, mac, name, manufacturer, model,
                channelCount, online, requiredInteger(object, "sampleIntervalMs"),
                requiredLong(object, "nextSequence"), optionalInstant(object, "lastSampledAt"),
                channels, latestMeasurements, history, collectorHistoryFaults(object.get("historyFaults")));
                channels, latestMeasurements, historyRecordCount,
                collectorHistoryFaults(object.get("historyFaults")));
    }

    private CollectorChannelState collectorChannelState(Object value)
@@ -270,4 +387,14 @@
        String value = object.getString(field);
        return value == null ? null : Instant.parse(value);
    }

    private String historyKey(String collectorCode)
    {
        return HISTORY_KEY_PREFIX + collectorCode;
    }

    private record ParsedSimulatorState(SimulatorState state, List<CollectorHistoryAppend> embeddedHistory)
    {
        private ParsedSimulatorState { embeddedHistory = List.copyOf(embeddedHistory); }
    }
}

 ruoyi-admin/src/test/java/com/ruoyi/config/RedisStateStoreTest.java

@@ -3,17 +3,26 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.api.Test;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.data.redis.core.SetOperations;
import org.springframework.data.redis.core.ValueOperations;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.device.model.DeviceWorkStatus;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelTombstone;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState;
@@ -78,7 +87,65 @@
        assertEquals(1L, collector.nextSequence());
        assertEquals(List.of("S-01", "S-02"),
                collector.channels().stream().map(channel -> channel.channelCode()).toList());
        assertEquals(List.of(), collector.history());
        assertEquals(0L, collector.historyRecordCount());
    }

    @Test
    void migratesEmbeddedCollectorHistoryToAppendOnlyRedisList()
    {
        AtomicReference<Object> stored = new AtomicReference<>("{\"version\":1,\"devices\":[],"
                + "\"collectors\":[{\"collectorId\":1,\"deviceCode\":\"dev-sim-collector-1\","
                + "\"ip\":\"127.0.1.2\",\"mac\":\"02:43:00:00:00:01\","
                + "\"name\":\"模拟采集器\",\"manufacturer\":\"开发模拟器\","
                + "\"model\":\"Virtual Collector\",\"channelCount\":1,\"online\":true,"
                + "\"sampleIntervalMs\":1000,\"nextSequence\":2,"
                + "\"lastSampledAt\":\"2026-08-29T05:00:00Z\","
                + "\"channels\":[{\"channelCode\":\"S-01\",\"metrics\":[{"
                + "\"metricCode\":\"temperature\",\"value\":25.1,\"unit\":\"℃\"}]}],"
                + "\"history\":[{\"adapterCode\":\"dev-sim\","
                + "\"collectorCode\":\"dev-sim-collector-1\",\"channelCode\":\"S-01\","
                + "\"sequenceId\":1,\"deviceTime\":\"2026-08-29T05:00:00Z\","
                + "\"deviceTimeRaw\":\"2026-08-29 13:00:00.000\","
                + "\"serverReceivedAt\":\"2026-08-29T05:00:00Z\","
                + "\"metricCode\":\"temperature\",\"value\":25.1,\"unit\":\"℃\"}],"
                + "\"historyFaults\":{\"missingSequences\":[],\"outOfOrder\":false}}],"
                + "\"nextDeviceId\":1,\"nextCollectorId\":2,\"tombstones\":[]}");
        RedisBtsSimulatorStateStore store = store(stored);

        SimulatorState restored = store.loadOrInitialize(state(1, "127.0.0.2"));

        assertEquals(1L, restored.collectors().get(0).historyRecordCount());
        assertEquals(1, store.loadCollectorHistory("dev-sim-collector-1").size());
        assertEquals(true, stored.get().toString().contains("\"history\":[{"));
        store.completeValidatedRestore(restored);
        assertEquals(false, stored.get().toString().contains("\"history\":[{"));
    }

    @Test
    void doesNotOverwriteDamagedEmbeddedHistoryBeforeValidation()
    {
        AtomicReference<Object> stored = new AtomicReference<>("{\"version\":1,\"devices\":[],"
                + "\"collectors\":[{\"collectorId\":1,\"deviceCode\":\"dev-sim-collector-1\","
                + "\"ip\":\"127.0.1.2\",\"mac\":\"02:43:00:00:00:01\","
                + "\"name\":\"模拟采集器\",\"manufacturer\":\"开发模拟器\","
                + "\"model\":\"Virtual Collector\",\"channelCount\":1,\"online\":true,"
                + "\"sampleIntervalMs\":1000,\"nextSequence\":2,"
                + "\"lastSampledAt\":\"2026-08-29T05:00:00Z\","
                + "\"channels\":[{\"channelCode\":\"S-01\",\"metrics\":[{"
                + "\"metricCode\":\"temperature\",\"value\":25.1,\"unit\":\"℃\"}]}],"
                + "\"history\":[{\"adapterCode\":\"dev-sim\","
                + "\"collectorCode\":\"wrong-collector\",\"channelCode\":\"S-01\","
                + "\"sequenceId\":1,\"deviceTime\":\"2026-08-29T05:00:00Z\","
                + "\"deviceTimeRaw\":\"2026-08-29 13:00:00.000\","
                + "\"serverReceivedAt\":\"2026-08-29T05:00:00Z\","
                + "\"metricCode\":\"temperature\",\"value\":25.1,\"unit\":\"℃\"}],"
                + "\"historyFaults\":{\"missingSequences\":[],\"outOfOrder\":false}}],"
                + "\"nextDeviceId\":1,\"nextCollectorId\":2,\"tombstones\":[]}");

        assertThrows(IllegalStateException.class,
                () -> new BtsDevelopmentSimulator(0, store(stored)));

        assertEquals(true, stored.get().toString().contains("\"history\":[{"));
    }

    @Test
@@ -93,6 +160,29 @@

        assertEquals("Unsupported BTS simulator state version: 2", exception.getMessage());
        assertEquals(true, stored.get().toString().contains("\"version\":2"));
    }

    @Test
    void atomicallySavesCollectorDeletionAndRemovesIndependentHistory()
    {
        AtomicReference<Object> stored = new AtomicReference<>();
        RedisBtsSimulatorStateStore store = store(stored);
        SimulatorState current = state(1, "127.0.0.2");
        store.loadOrInitialize(current);
        var record = new com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord(
                "dev-sim", "dev-sim-collector-1", "S-01", 1L,
                java.time.Instant.parse("2026-08-29T05:00:00Z"), "2026-08-29 13:00:00.000",
                java.time.Instant.parse("2026-08-29T05:00:00Z"), "temperature",
                new java.math.BigDecimal("25.1"), "℃");
        store.save(current, List.of(
                new com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryAppend(
                        "dev-sim-collector-1", List.of(record))));
        SimulatorState next = state(7, "127.0.0.9");

        store.saveAndDeleteCollectorHistory(next, "dev-sim-collector-1");

        assertEquals(next, store.loadOrInitialize(current));
        assertEquals(List.of(), store.loadCollectorHistory("dev-sim-collector-1"));
    }

    @Test
@@ -175,7 +265,37 @@
    {
        RedisTemplate<Object, Object> redis = mock(RedisTemplate.class);
        ValueOperations<Object, Object> values = mock(ValueOperations.class);
        ListOperations<Object, Object> lists = mock(ListOperations.class);
        SetOperations<Object, Object> sets = mock(SetOperations.class);
        Map<Object, List<Object>> listValues = new LinkedHashMap<>();
        when(redis.opsForValue()).thenReturn(values);
        when(redis.opsForList()).thenReturn(lists);
        when(redis.opsForSet()).thenReturn(sets);
        when(sets.add(any(), any())).thenReturn(1L);
        when(sets.members(any())).thenReturn(Set.of());
        when(redis.exec()).thenReturn(List.of(Boolean.TRUE));
        when(redis.execute(any(SessionCallback.class))).thenAnswer(invocation -> {
            SessionCallback<?> callback = invocation.getArgument(0);
            return callback.execute(redis);
        });
        when(lists.rightPushAll(any(), org.mockito.ArgumentMatchers.<Object>anyCollection()))
                .thenAnswer(invocation -> {
                    Object key = invocation.getArgument(0);
                    List<Object> records = new ArrayList<>(listValues.getOrDefault(key, List.of()));
                    FastJson2JsonRedisSerializer<Object> serializer =
                            new FastJson2JsonRedisSerializer<>(Object.class);
                    List<Object> appended = invocation.getArgument(1);
                    appended.forEach(item -> records.add(
                            serializer.deserialize(serializer.serialize(item))));
                    listValues.put(key, records);
                    return (long) records.size();
                });
        when(lists.range(any(), anyLong(), anyLong())).thenAnswer(invocation ->
                List.copyOf(listValues.getOrDefault(invocation.getArgument(0), List.of())));
        when(lists.size(any())).thenAnswer(invocation ->
                (long) listValues.getOrDefault(invocation.getArgument(0), List.of()).size());
        when(redis.delete(org.mockito.ArgumentMatchers.<Object>any()))
                .thenAnswer(invocation -> listValues.remove(invocation.getArgument(0)) != null);
        when(values.get(any())).thenAnswer(invocation -> stored.get());
        when(values.setIfAbsent(any(), any())).thenAnswer(invocation -> {
            if (stored.get() != null) return false;

 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

 ruoyi-device/src/main/java/com/ruoyi/device/simulator/BtsSimulatorStateStore.java

@@ -16,6 +16,29 @@
    /** 完整保存下一版模拟器状态。 */
    void save(SimulatorState state);

    /** 状态和独立历史全部校验通过后，提交兼容旧格式所需的迁移。 */
    default void completeValidatedRestore(SimulatorState state) { }

    /** 原子保存主状态并追加本轮新产生的原始历史。 */
    default void save(SimulatorState state, List<CollectorHistoryAppend> historyAppends)
    {
        if (!historyAppends.isEmpty())
            throw new UnsupportedOperationException("Collector history append is not supported");
        save(state);
    }

    /** 读取指定采集器的只追加原始历史。 */
    default List<CollectorMeasurementRecord> loadCollectorHistory(String collectorCode)
    {
        return List.of();
    }

    /** 原子保存删除采集器后的主状态，并删除该采集器的独立历史。 */
    default void saveAndDeleteCollectorHistory(SimulatorState state, String collectorCode)
    {
        throw new UnsupportedOperationException("Atomic collector deletion is not supported");
    }

    /** 读取短期保留的最近请求。 */
    List<RecentRequest> loadRecentRequests();

@@ -47,13 +70,12 @@
            String manufacturer, String model, int channelCount, boolean online, int sampleIntervalMs,
            long nextSequence, Instant lastSampledAt, List<CollectorChannelState> channels,
            List<CollectorRealtimeMeasurement> latestMeasurements,
            List<CollectorMeasurementRecord> history, CollectorHistoryFaults historyFaults)
            long historyRecordCount, CollectorHistoryFaults historyFaults)
    {
        public CollectorState
        {
            channels = List.copyOf(channels);
            latestMeasurements = List.copyOf(latestMeasurements);
            history = List.copyOf(history);
            historyFaults = historyFaults == null ? CollectorHistoryFaults.none() : historyFaults;
        }

@@ -63,7 +85,7 @@
        {
            this(collectorId, deviceCode, ip, mac, name, manufacturer, model, channelCount, online,
                    DevelopmentCollectorSimulation.DEFAULT_SAMPLE_INTERVAL_MS, 1L, null,
                    DevelopmentCollectorSimulation.defaultChannels(channelCount), List.of(), List.of(),
                    DevelopmentCollectorSimulation.defaultChannels(channelCount), List.of(), 0L,
                    CollectorHistoryFaults.none());
        }
    }
@@ -83,6 +105,11 @@
            long sequenceId, Instant deviceTime, String deviceTimeRaw, Instant serverReceivedAt,
            String metricCode, BigDecimal value, String unit) { }

    record CollectorHistoryAppend(String collectorCode, List<CollectorMeasurementRecord> records)
    {
        public CollectorHistoryAppend { records = List.copyOf(records); }
    }

    record CollectorHistoryFaults(List<Long> missingSequences, Long duplicateSequence,
            Long conflictingSequence, boolean outOfOrder)
    {

 ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java

@@ -61,7 +61,7 @@
        List<CollectorRealtimeMeasurement> latestMeasurements = retainConfiguredMeasurements(
                current.latestMeasurements(), channels);
        return copy(current, ip, channelCount, online, current.sampleIntervalMs(), current.nextSequence(),
                lastSampledAt, channels, latestMeasurements, current.history(), current.historyFaults());
                lastSampledAt, channels, latestMeasurements, current.historyRecordCount(), current.historyFaults());
    }

    /** 更新采样频率和历史下载故障；测量原始历史保持只追加。 */
@@ -72,7 +72,7 @@
        validateFaults(historyFaults);
        return copy(current, current.ip(), current.channelCount(), current.online(), sampleIntervalMs,
                current.nextSequence(), current.lastSampledAt(), current.channels(), current.latestMeasurements(),
                current.history(), historyFaults);
                current.historyRecordCount(), historyFaults);
    }

    /** 替换一个演示通道的指标配置；动态添加项与固定指标共用指标编码合同。 */
@@ -87,19 +87,19 @@
                current.latestMeasurements(), channels);
        return copy(current, current.ip(), current.channelCount(), current.online(), current.sampleIntervalMs(),
                current.nextSequence(), current.lastSampledAt(), channels, latestMeasurements,
                current.history(), current.historyFaults());
                current.historyRecordCount(), current.historyFaults());
    }

    /** 按采集器自身频率推进采样；离线期间不生成记录，恢复后从下一序号继续。 */
    public static CollectorState advance(CollectorState current, Instant now)
    public static CollectorAdvance advance(CollectorState current, Instant now)
    {
        if (!current.online()) return current;
        if (!current.online()) return new CollectorAdvance(current, List.of());
        Instant sampleAt = current.lastSampledAt() == null
                ? now
                : current.lastSampledAt().plusMillis(current.sampleIntervalMs());
        if (sampleAt.isAfter(now)) return current;
        if (sampleAt.isAfter(now)) return new CollectorAdvance(current, List.of());

        List<CollectorMeasurementRecord> history = new ArrayList<>(current.history());
        List<CollectorMeasurementRecord> appendedRecords = new ArrayList<>();
        Map<String, CollectorRealtimeMeasurement> latest = new LinkedHashMap<>();
        for (CollectorRealtimeMeasurement measurement : current.latestMeasurements())
            latest.put(measurementKey(measurement.channelCode(), measurement.metricCode()), measurement);
@@ -116,7 +116,7 @@
                            channel.channelCode(), sequence, sampleAt,
                            DEVICE_TIME_FORMATTER.format(sampleAt.atZone(DEVICE_ZONE)), now,
                            metric.metricCode(), metric.value(), metric.unit());
                    history.add(record);
                    appendedRecords.add(record);
                    latest.put(measurementKey(record.channelCode(), record.metricCode()),
                            new CollectorRealtimeMeasurement(record.adapterCode(), record.collectorCode(),
                                    record.channelCode(), record.sequenceId(), record.deviceTime(),
@@ -129,9 +129,12 @@
            generated++;
            sampleAt = sampleAt.plusMillis(current.sampleIntervalMs());
        }
        return copy(current, current.ip(), current.channelCount(), current.online(), current.sampleIntervalMs(),
                sequence, lastSampledAt, current.channels(), List.copyOf(latest.values()), history,
        CollectorState next = copy(current, current.ip(), current.channelCount(), current.online(),
                current.sampleIntervalMs(),
                sequence, lastSampledAt, current.channels(), List.copyOf(latest.values()),
                current.historyRecordCount() + appendedRecords.size(),
                current.historyFaults());
        return new CollectorAdvance(next, appendedRecords);
    }

    /** 返回独立保存的最新实时快照，不从原始历史反推当前状态。 */
@@ -142,13 +145,14 @@
    }

    /** 对只追加原始模拟历史应用可重复的下载故障视图，再按偏移量分页。 */
    public static CollectorHistoryPage history(CollectorState current, long offset, int limit)
    public static CollectorHistoryPage history(CollectorState current, List<CollectorMeasurementRecord> rawHistory,
            long offset, int limit)
    {
        if (offset < 0) throw new IllegalArgumentException("offset must not be negative");
        if (limit < 1 || limit > MAX_HISTORY_PAGE_SIZE)
            throw new IllegalArgumentException("limit must be between 1 and " + MAX_HISTORY_PAGE_SIZE);

        List<CollectorMeasurementRecord> download = faultedHistory(current);
        List<CollectorMeasurementRecord> download = faultedHistory(current, rawHistory);
        int from = (int) Math.min(offset, download.size());
        int to = Math.min(from + limit, download.size());
        return new CollectorHistoryPage(current.deviceCode(), offset, to, to < download.size(),
@@ -156,7 +160,7 @@
    }

    /** 校验 Redis 恢复的采集器采样状态，拒绝损坏状态被静默替换。 */
    public static void validate(CollectorState state)
    public static void validate(CollectorState state, List<CollectorMeasurementRecord> rawHistory)
    {
        validateSampleInterval(state.sampleIntervalMs());
        if (state.nextSequence() < 1 || state.channels().size() != state.channelCount())
@@ -169,7 +173,7 @@
            validateMetrics(channel.metrics());
        }
        validateLatestMeasurements(state);
        validateHistory(state);
        validateHistory(state, rawHistory);
        validateFaults(state.historyFaults());
    }

@@ -189,12 +193,12 @@
        }
    }

    private static void validateHistory(CollectorState state)
    private static void validateHistory(CollectorState state, List<CollectorMeasurementRecord> rawHistory)
    {
        long maximumSequence = 0;
        long previousSequence = 0;
        Set<String> identities = new java.util.HashSet<>();
        for (CollectorMeasurementRecord record : state.history())
        for (CollectorMeasurementRecord record : rawHistory)
        {
            String identity = record.sequenceId() + "\u0000" + record.channelCode() + "\u0000" + record.metricCode();
            if (!validMeasurement(state, record.adapterCode(), record.collectorCode(), record.channelCode(),
@@ -207,6 +211,8 @@
        }
        if (state.nextSequence() <= maximumSequence)
            throw new IllegalStateException("Invalid collector history sequence in Redis");
        if (state.historyRecordCount() != rawHistory.size())
            throw new IllegalStateException("Invalid collector history count in Redis");
    }

    private static boolean validMeasurement(CollectorState state, String adapterCode, String collectorCode,
@@ -266,12 +272,13 @@
        return channelCode + "\u0000" + metricCode;
    }

    private static List<CollectorMeasurementRecord> faultedHistory(CollectorState current)
    private static List<CollectorMeasurementRecord> faultedHistory(CollectorState current,
            List<CollectorMeasurementRecord> rawHistory)
    {
        CollectorHistoryFaults faults = current.historyFaults();
        Set<Long> missing = Set.copyOf(faults.missingSequences());
        List<CollectorMeasurementRecord> records = new ArrayList<>();
        for (CollectorMeasurementRecord record : current.history())
        for (CollectorMeasurementRecord record : rawHistory)
        {
            if (missing.contains(record.sequenceId())) continue;
            records.add(record);
@@ -292,12 +299,12 @@
    private static CollectorState copy(CollectorState current, String ip, int channelCount, boolean online,
            int sampleIntervalMs, long nextSequence, Instant lastSampledAt,
            List<CollectorChannelState> channels, List<CollectorRealtimeMeasurement> latestMeasurements,
            List<CollectorMeasurementRecord> history,
            long historyRecordCount,
            CollectorHistoryFaults faults)
    {
        return new CollectorState(current.collectorId(), current.deviceCode(), ip.trim(), current.mac(),
                current.name(), current.manufacturer(), current.model(), channelCount, online, sampleIntervalMs,
                nextSequence, lastSampledAt, channels, latestMeasurements, history, faults);
                nextSequence, lastSampledAt, channels, latestMeasurements, historyRecordCount, faults);
    }

    private static void validateSampleInterval(int sampleIntervalMs)
@@ -315,7 +322,9 @@
        for (CollectorMetricState metric : metrics)
        {
            if (metric.metricCode() == null || metric.metricCode().isBlank() || metric.value() == null
                    || metric.unit() == null || metric.unit().isBlank() || !codes.add(metric.metricCode().trim()))
                    || !metric.metricCode().equals(metric.metricCode().trim())
                    || metric.unit() == null || metric.unit().isBlank()
                    || !metric.unit().equals(metric.unit().trim()) || !codes.add(metric.metricCode()))
                throw new IllegalArgumentException("Collector metric code, value and unit must be valid and unique");
        }
    }
@@ -350,6 +359,11 @@
        public CollectorRealtimeSnapshot { measurements = List.copyOf(measurements); }
    }

    public record CollectorAdvance(CollectorState state, List<CollectorMeasurementRecord> appendedRecords)
    {
        public CollectorAdvance { appendedRecords = List.copyOf(appendedRecords); }
    }

    public record CollectorHistoryPage(String collectorCode, long offset, long nextOffset, boolean hasMore,
            long total, List<CollectorMeasurementRecord> records)
    {

 ruoyi-device/src/main/java/com/ruoyi/device/simulator/InMemoryBtsSimulatorStateStore.java

@@ -1,6 +1,11 @@
package com.ruoyi.device.simulator;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryAppend;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.RecentRequest;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.SimulatorState;

@@ -9,6 +14,7 @@
{
    private SimulatorState state;
    private List<RecentRequest> recentRequests = List.of();
    private final Map<String, List<CollectorMeasurementRecord>> collectorHistory = new LinkedHashMap<>();

    public InMemoryBtsSimulatorStateStore() { }

@@ -31,6 +37,32 @@
    }

    @Override
    public synchronized void save(SimulatorState state, List<CollectorHistoryAppend> historyAppends)
    {
        for (CollectorHistoryAppend append : historyAppends)
        {
            List<CollectorMeasurementRecord> records = new ArrayList<>(
                    collectorHistory.getOrDefault(append.collectorCode(), List.of()));
            records.addAll(append.records());
            collectorHistory.put(append.collectorCode(), List.copyOf(records));
        }
        this.state = state;
    }

    @Override
    public synchronized List<CollectorMeasurementRecord> loadCollectorHistory(String collectorCode)
    {
        return collectorHistory.getOrDefault(collectorCode, List.of());
    }

    @Override
    public synchronized void saveAndDeleteCollectorHistory(SimulatorState state, String collectorCode)
    {
        collectorHistory.remove(collectorCode);
        this.state = state;
    }

    @Override
    public synchronized List<RecentRequest> loadRecentRequests()
    {
        return recentRequests;
@@ -47,5 +79,6 @@
    {
        state = defaultState;
        recentRequests = List.of();
        collectorHistory.clear();
    }
}

 ruoyi-device/src/test/java/com/ruoyi/device/contract/SimulatorPublicContractTest.java

@@ -11,6 +11,7 @@
import java.util.List;
import java.util.Set;
import org.junit.jupiter.api.Test;
import com.alibaba.fastjson2.JSON;
import com.ruoyi.device.api.DeviceAdapter;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.bts.BtsDeviceAdapter;
@@ -22,6 +23,7 @@
import com.ruoyi.device.simulator.BtsSimulatorStateStore;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelState;
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;
@@ -132,6 +134,20 @@
    }

    @Test
    void collectorRawHistoryIsNotEmbeddedInMainSimulatorState() throws Exception
    {
        InMemoryBtsSimulatorStateStore store = new InMemoryBtsSimulatorStateStore();
        try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0, store))
        {
            simulator.advanceCollectors(Instant.parse("2026-08-29T03:30:00Z"));

            String mainState = JSON.toJSONString(store.loadOrInitialize(null));
            assertFalse(mainState.contains("\"history\":[{"));
            assertEquals(12, simulator.collectorHistory(1, 0, 100).total());
        }
    }

    @Test
    void collectorRealtimeDropsMeasurementsRemovedFromCurrentConfiguration() throws Exception
    {
        Instant startedAt = Instant.parse("2026-08-29T04:00:00Z");
@@ -159,6 +175,18 @@
        {
            assertEquals(4, restored.snapshot().collectors().get(0).latestMeasurements().size());
            assertEquals(12, restored.collectorHistory(1, 0, 100).total());
        }
    }

    @Test
    void collectorRejectsMetricWhitespaceThatWouldBreakRestore() throws Exception
    {
        try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0))
        {
            assertThrows(IllegalArgumentException.class, () -> simulator.configureCollectorChannel(1, "S-01",
                    List.of(new CollectorMetricState("temperature ", new BigDecimal("25.0"), "℃"))));
            assertThrows(IllegalArgumentException.class, () -> simulator.configureCollectorChannel(1, "S-01",
                    List.of(new CollectorMetricState("temperature", new BigDecimal("25.0"), " ℃"))));
        }
    }

@@ -347,6 +375,31 @@
    }

    @Test
    void collectorDeletionDoesNotReplaceMemoryWhenAtomicPersistenceFails() throws Exception
    {
        InMemoryBtsSimulatorStateStore delegate = new InMemoryBtsSimulatorStateStore();
        BtsSimulatorStateStore failing = new BtsSimulatorStateStore()
        {
            @Override public SimulatorState loadOrInitialize(SimulatorState value)
            { return delegate.loadOrInitialize(value); }
            @Override public void save(SimulatorState value) { delegate.save(value); }
            @Override public void saveAndDeleteCollectorHistory(SimulatorState value, String collectorCode)
            { throw new IllegalStateException("redis unavailable"); }
            @Override public java.util.List<RecentRequest> loadRecentRequests()
            { return delegate.loadRecentRequests(); }
            @Override public void saveRecentRequests(java.util.List<RecentRequest> requests)
            { delegate.saveRecentRequests(requests); }
            @Override public void reset(SimulatorState value) { delegate.reset(value); }
        };

        try (BtsDevelopmentSimulator simulator = new BtsDevelopmentSimulator(0, failing))
        {
            assertThrows(IllegalStateException.class, () -> simulator.deleteCollector(1));
            assertEquals(1, simulator.snapshot().collectors().size());
        }
    }

    @Test
    void developmentSimulatorRestoresUnfinishedTombstones() throws Exception
    {
        InMemoryBtsSimulatorStateStore store = new InMemoryBtsSimulatorStateStore();
@@ -390,17 +443,20 @@
        Instant observedAt = Instant.parse("2026-08-29T05:00:00Z");
        List<CollectorChannelState> channels = List.of(new CollectorChannelState("S-01", List.of(
                new CollectorMetricState("temperature", new BigDecimal("25.0"), "℃"))));
        CollectorState collector = new CollectorState(1, "dev-sim-collector-1", "127.0.1.2",
                "02:43:00:00:00:01", "模拟采集器", "开发模拟器", "Virtual Collector", 1, true,
                1000, 2L, observedAt, channels, List.of(), List.of(new CollectorMeasurementRecord(
        CollectorMeasurementRecord record = new CollectorMeasurementRecord(
                        "dev-sim", "wrong-collector", "S-01", 1L, observedAt,
                        "2026-08-29 13:00:00.000", observedAt, "temperature",
                        new BigDecimal("25.0"), "℃")), CollectorHistoryFaults.none());
                new BigDecimal("25.0"), "℃");
        CollectorState collector = new CollectorState(1, "dev-sim-collector-1", "127.0.1.2",
                "02:43:00:00:00:01", "模拟采集器", "开发模拟器", "Virtual Collector", 1, true,
                1000, 2L, observedAt, channels, List.of(), 1L, CollectorHistoryFaults.none());
        SimulatorState damaged = new SimulatorState(BtsSimulatorStateStore.CURRENT_VERSION,
                List.of(), List.of(collector), 1, 2, List.of());
        InMemoryBtsSimulatorStateStore store = new InMemoryBtsSimulatorStateStore(damaged);
        store.save(damaged, List.of(new CollectorHistoryAppend(collector.deviceCode(), List.of(record))));

        assertThrows(IllegalStateException.class,
                () -> new BtsDevelopmentSimulator(0, new InMemoryBtsSimulatorStateStore(damaged)));
                () -> new BtsDevelopmentSimulator(0, store));
    }

    @Test
@@ -409,16 +465,19 @@
        Instant observedAt = Instant.parse("2026-08-29T06:00:00Z");
        List<CollectorChannelState> channels = List.of(new CollectorChannelState("S-01", List.of(
                new CollectorMetricState("temperature", new BigDecimal("25.0"), "℃"))));
        CollectorState collector = new CollectorState(1, "dev-sim-collector-1", "127.0.1.2",
                "02:43:00:00:00:01", "模拟采集器", "开发模拟器", "Virtual Collector", 1, true,
                1000, 1L, observedAt, channels, List.of(), List.of(new CollectorMeasurementRecord(
        CollectorMeasurementRecord record = new CollectorMeasurementRecord(
                        "dev-sim", "dev-sim-collector-1", "S-01", 1L, observedAt,
                        "2026-08-29 14:00:00.000", observedAt, "temperature",
                        new BigDecimal("25.0"), "℃")), CollectorHistoryFaults.none());
                new BigDecimal("25.0"), "℃");
        CollectorState collector = new CollectorState(1, "dev-sim-collector-1", "127.0.1.2",
                "02:43:00:00:00:01", "模拟采集器", "开发模拟器", "Virtual Collector", 1, true,
                1000, 1L, observedAt, channels, List.of(), 1L, CollectorHistoryFaults.none());
        SimulatorState damaged = new SimulatorState(BtsSimulatorStateStore.CURRENT_VERSION,
                List.of(), List.of(collector), 1, 2, List.of());
        InMemoryBtsSimulatorStateStore store = new InMemoryBtsSimulatorStateStore(damaged);
        store.save(damaged, List.of(new CollectorHistoryAppend(collector.deviceCode(), List.of(record))));

        assertThrows(IllegalStateException.class,
                () -> new BtsDevelopmentSimulator(0, new InMemoryBtsSimulatorStateStore(damaged)));
                () -> new BtsDevelopmentSimulator(0, store));
    }
}