From 370ef4f5e663fdfc1d19d2774dfc4ec1ed0af60e Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 29 八月 2026 14:57:48 +0800
Subject: [PATCH] 修复采集器历史状态持续序列化错误
---
ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java | 151 ++++++++++++++++++++++++++++++++++++++++++++++----
1 files changed, 139 insertions(+), 12 deletions(-)
diff --git a/ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java b/ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java
index 0b9bb63..6c9d1eb 100644
--- a/ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java
+++ b/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); }
+ }
}
--
Gitblit v1.7.1