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

yeg

2026-08-29 370ef4f5e663fdfc1d19d2774dfc4ec1ed0af60e

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

@@ -1,8 +1,13 @@
package com.ruoyi.config;

import java.math.BigDecimal;
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;
@@ -17,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)
    {
@@ -30,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
@@ -62,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
        {
@@ -78,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)
        {
@@ -130,13 +240,116 @@
                object.getBooleanValue("online"), channels);
    }

    private CollectorState collectorState(Object value)
    private List<CollectorMeasurementRecord> embeddedCollectorHistory(JSONObject object)
    {
        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");
        String mac = object.getString("mac");
        String name = object.getString("name");
        String manufacturer = object.getString("manufacturer");
        String model = object.getString("model");
        int channelCount = requiredInteger(object, "channelCount");
        boolean online = object.getBooleanValue("online");
        // Ticket04 状态只包含发现配置；首次恢复时以演示默认值补齐采样状态。
        if (!object.containsKey("sampleIntervalMs"))
            return new CollectorState(collectorId, deviceCode, ip, mac, name, manufacturer, model,
                    channelCount, online);

        List<CollectorChannelState> channels = new ArrayList<>();
        requiredArray(object, "channels").forEach(item -> channels.add(collectorChannelState(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(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, historyRecordCount,
                collectorHistoryFaults(object.get("historyFaults")));
    }

    private CollectorChannelState collectorChannelState(Object value)
    {
        JSONObject object = jsonObject(value);
        return new CollectorState(requiredInteger(object, "collectorId"), object.getString("deviceCode"),
                object.getString("ip"), object.getString("mac"), object.getString("name"),
                object.getString("manufacturer"), object.getString("model"),
                requiredInteger(object, "channelCount"), object.getBooleanValue("online"));
        List<CollectorMetricState> metrics = new ArrayList<>();
        requiredArray(object, "metrics").forEach(item -> metrics.add(collectorMetricState(item)));
        return new CollectorChannelState(object.getString("channelCode"), metrics);
    }

    private CollectorMetricState collectorMetricState(Object value)
    {
        JSONObject object = jsonObject(value);
        BigDecimal measurement = object.getBigDecimal("value");
        if (measurement == null) throw new IllegalStateException("Missing BTS simulator state field: value");
        return new CollectorMetricState(object.getString("metricCode"), measurement, object.getString("unit"));
    }

    private CollectorMeasurementRecord collectorMeasurementRecord(Object value)
    {
        JSONObject object = jsonObject(value);
        BigDecimal measurement = object.getBigDecimal("value");
        if (measurement == null) throw new IllegalStateException("Missing BTS simulator state field: value");
        return new CollectorMeasurementRecord(object.getString("adapterCode"), object.getString("collectorCode"),
                object.getString("channelCode"), requiredLong(object, "sequenceId"),
                requiredInstant(object, "deviceTime"), object.getString("deviceTimeRaw"),
                requiredInstant(object, "serverReceivedAt"), object.getString("metricCode"),
                measurement, object.getString("unit"));
    }

    private CollectorRealtimeMeasurement collectorRealtimeMeasurement(Object value)
    {
        JSONObject object = jsonObject(value);
        BigDecimal measurement = object.getBigDecimal("value");
        if (measurement == null) throw new IllegalStateException("Missing BTS simulator state field: value");
        return new CollectorRealtimeMeasurement(object.getString("adapterCode"), object.getString("collectorCode"),
                object.getString("channelCode"), requiredLong(object, "sequenceId"),
                requiredInstant(object, "deviceTime"), object.getString("deviceTimeRaw"),
                requiredInstant(object, "serverReceivedAt"), object.getString("metricCode"),
                measurement, object.getString("unit"));
    }

    /** 兼容 Ticket06 开发期间已写入但尚未拆分快照模型的状态。 */
    private List<CollectorRealtimeMeasurement> latestMeasurementsFromLegacyHistory(
            List<CollectorMeasurementRecord> history, List<CollectorChannelState> channels)
    {
        java.util.Set<String> configured = new java.util.HashSet<>();
        channels.forEach(channel -> channel.metrics().forEach(metric ->
                configured.add(channel.channelCode() + "\u0000" + metric.metricCode())));
        java.util.Map<String, CollectorRealtimeMeasurement> latest = new java.util.LinkedHashMap<>();
        for (CollectorMeasurementRecord record : history)
        {
            String key = record.channelCode() + "\u0000" + record.metricCode();
            if (configured.contains(key))
                latest.put(key, new CollectorRealtimeMeasurement(record.adapterCode(), record.collectorCode(),
                        record.channelCode(), record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(),
                        record.serverReceivedAt(), record.metricCode(), record.value(), record.unit()));
        }
        return List.copyOf(latest.values());
    }

    private CollectorHistoryFaults collectorHistoryFaults(Object value)
    {
        if (value == null) return CollectorHistoryFaults.none();
        JSONObject object = jsonObject(value);
        List<Long> missing = new ArrayList<>();
        JSONArray values = object.getJSONArray("missingSequences");
        if (values != null) values.forEach(item -> missing.add(Long.valueOf(item.toString())));
        return new CollectorHistoryFaults(missing, object.getLong("duplicateSequence"),
                object.getLong("conflictingSequence"), object.getBooleanValue("outOfOrder"));
    }

    private ChannelState channelState(Object value)
@@ -154,4 +367,34 @@
        return new ChannelTombstone(requiredInteger(object, "deviceId"), object.getString("ip"),
                requiredInteger(object, "channelId"), requiredInteger(object, "remainingDiscoveries"));
    }

    private long requiredLong(JSONObject object, String field)
    {
        Long value = object.getLong(field);
        if (value == null) throw new IllegalStateException("Missing BTS simulator state field: " + field);
        return value;
    }

    private Instant requiredInstant(JSONObject object, String field)
    {
        Instant value = optionalInstant(object, field);
        if (value == null) throw new IllegalStateException("Missing BTS simulator state field: " + field);
        return value;
    }

    private Instant optionalInstant(JSONObject object, String field)
    {
        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); }
    }
}