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

yeg

2026-08-29 370ef4f5e663fdfc1d19d2774dfc4ec1ed0af60e

 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); }
    }
}