From 7259e265b9237286996b09da03038ea5a133e1e0 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 05 九月 2026 17:45:07 +0800
Subject: [PATCH] 收敛模拟器为设备事实与实时观测
---
ruoyi-admin/src/main/java/com/battery/business/runtime/RedisBtsSimulatorStateStore.java | 404 ++-------------------------------------------------------
1 files changed, 19 insertions(+), 385 deletions(-)
diff --git a/ruoyi-admin/src/main/java/com/battery/business/runtime/RedisBtsSimulatorStateStore.java b/ruoyi-admin/src/main/java/com/battery/business/runtime/RedisBtsSimulatorStateStore.java
index f407c87..8eed74a 100644
--- a/ruoyi-admin/src/main/java/com/battery/business/runtime/RedisBtsSimulatorStateStore.java
+++ b/ruoyi-admin/src/main/java/com/battery/business/runtime/RedisBtsSimulatorStateStore.java
@@ -1,13 +1,9 @@
package com.battery.business.runtime;
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 org.springframework.context.annotation.Profile;
@@ -16,9 +12,6 @@
import com.alibaba.fastjson2.JSONObject;
import com.battery.device.simulator.BtsSimulatorStateStore;
import com.battery.device.simulator.BtsSimulatorStateStore.BtsDeviceProfile;
-import com.battery.device.simulator.BtsSimulatorStateStore.BtsHistoryBatch;
-import com.battery.device.simulator.BtsSimulatorStateStore.BtsHistoryRecord;
-import com.battery.device.simulator.BtsSimulatorStateStore.BtsPlaybackState;
import com.battery.device.api.model.DeviceWorkStatus;
/** 将开发模拟器完整状态写入 Redis，主状态不设置普通 TTL。 */
@@ -28,16 +21,9 @@
{
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:";
- // 历史与播放位置使用独立键版本，避免把旧的三循环演示历史恢复到当前时间轴。
- static final String BTS_HISTORY_KEYS_KEY = "dev:bts-simulator:bts-history-keys:v2";
- static final String BTS_HISTORY_KEY_PREFIX = "dev:bts-simulator:bts-history:v2:";
- static final String BTS_PLAYBACK_KEY_PREFIX = "dev:bts-simulator:bts-playback:v2:";
private static final long RECENT_REQUESTS_TTL_HOURS = 1;
private final RedisTemplate<Object, Object> redisTemplate;
- private List<CollectorHistoryAppend> pendingEmbeddedHistory = List.of();
public RedisBtsSimulatorStateStore(RedisTemplate<Object, Object> redisTemplate)
{
@@ -57,236 +43,9 @@
}
@Override
- public SimulatorState resetHistoryForDemoStartup(SimulatorState state)
- {
- if (pendingEmbeddedHistory.isEmpty()) return state;
- java.util.Map<String, Integer> embeddedCounts = new java.util.HashMap<>();
- pendingEmbeddedHistory.forEach(append -> embeddedCounts.put(append.collectorCode(), append.records().size()));
- List<CollectorState> collectors = state.collectors().stream().map(collector -> {
- int embeddedCount = embeddedCounts.getOrDefault(collector.deviceCode(), 0);
- if (embeddedCount == 0) return collector;
- long remaining = Math.max(0L, collector.historyRecordCount() - embeddedCount);
- return new CollectorState(collector.collectorId(), collector.deviceCode(), collector.ip(), collector.mac(),
- collector.name(), collector.manufacturer(), collector.model(), collector.channelCount(),
- collector.online(), collector.sampleIntervalMs(), collector.nextSequence(), collector.lastSampledAt(),
- collector.channels(), collector.latestMeasurements().stream()
- .filter(item -> collector.deviceCode().equals(item.collectorCode()))
- .toList(), remaining, collector.historyFaults());
- }).toList();
- // 不删除独立历史 Redis key；这里只阻止旧内嵌数据在本次启动时迁移回来。
- pendingEmbeddedHistory = List.of();
- SimulatorState cleaned = new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
- state.nextCollectorId(), state.tombstones());
- save(cleaned);
- return cleaned;
- }
-
- 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)
{
- 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.opsForList().trim(historyKey, -MAX_HISTORY_RECORDS, -1);
- 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 CollectorHistoryBatch loadCollectorHistoryPage(String collectorCode, long offset, int limit,
- boolean reverse)
- {
- if (offset < 0) throw new IllegalArgumentException("offset must not be negative");
- if (limit < 1 || limit > MAX_HISTORY_BATCH_SIZE)
- throw new IllegalArgumentException("limit must be between 1 and " + MAX_HISTORY_BATCH_SIZE);
- String key = historyKey(collectorCode);
- Long storedSize = redisTemplate.opsForList().size(key);
- long redisCount = storedSize == null ? 0 : storedSize;
- List<CollectorMeasurementRecord> embedded = pendingEmbeddedHistory.stream()
- .filter(append -> append.collectorCode().equals(collectorCode))
- .findFirst().map(CollectorHistoryAppend::records).orElse(List.of());
- long total = redisCount + embedded.size();
- long from = Math.min(offset, total);
- int count = (int) Math.min(limit, total - from);
- List<CollectorMeasurementRecord> records = new ArrayList<>(count);
- if (reverse)
- {
- long lower = total - from - count;
- long upper = total - from;
- long embeddedLower = Math.max(lower, redisCount);
- for (long index = upper - 1; index >= embeddedLower; index--)
- records.add(embedded.get((int) (index - redisCount)));
- long redisUpper = Math.min(upper, redisCount);
- if (lower < redisUpper)
- {
- List<Object> stored = redisTemplate.opsForList().range(key, lower, redisUpper - 1);
- if (stored != null)
- for (int index = stored.size() - 1; index >= 0; index--)
- records.add(collectorMeasurementRecord(stored.get(index)));
- }
- }
- else
- {
- long upper = from + count;
- long redisUpper = Math.min(upper, redisCount);
- if (from < redisUpper)
- {
- List<Object> stored = redisTemplate.opsForList().range(key, from, redisUpper - 1);
- if (stored != null) stored.forEach(item -> records.add(collectorMeasurementRecord(item)));
- }
- long embeddedFrom = Math.max(from, redisCount);
- if (embeddedFrom < upper)
- records.addAll(embedded.subList((int) (embeddedFrom - redisCount),
- (int) (upper - redisCount)));
- }
- if (records.size() != count)
- throw new IllegalStateException("Collector history page changed while reading Redis");
- long nextOffset = from + records.size();
- return new CollectorHistoryBatch(offset, nextOffset, nextOffset < total, total, records);
- }
-
- @Override
- public BtsHistoryBatch loadBtsHistoryPage(String historyKey, long offset, int limit)
- {
- if (offset < 0 || limit < 1 || limit > MAX_HISTORY_BATCH_SIZE)
- throw new IllegalArgumentException("invalid BTS history page");
- String key = BTS_HISTORY_KEY_PREFIX + historyKey;
- Long size = redisTemplate.opsForList().size(key);
- long total = size == null ? 0 : size;
- long from = Math.min(offset, total);
- int count = (int) Math.min(limit, total - from);
- List<Object> values = count == 0 ? List.of() : redisTemplate.opsForList().range(key, from, from + count - 1);
- List<BtsHistoryRecord> records = values == null ? List.of() : values.stream()
- .map(item -> JSON.parseObject(item.toString(), BtsHistoryRecord.class)).toList();
- long next = from + records.size();
- return new BtsHistoryBatch(offset, next, next < total, total, records);
- }
-
- @Override
- @SuppressWarnings({ "rawtypes", "unchecked" })
- public void appendBtsHistory(String historyKey, BtsHistoryRecord record)
- {
- String key = BTS_HISTORY_KEY_PREFIX + historyKey;
- RedisOperations<Object, Object> operations = redisTemplate;
- operations.opsForList().rightPush(key, JSON.toJSONString(record));
- operations.opsForList().trim(key, -MAX_HISTORY_RECORDS, -1);
- operations.opsForSet().add(BTS_HISTORY_KEYS_KEY, key);
- }
-
- @Override
- public BtsPlaybackState loadBtsPlaybackState(String historyKey)
- {
- Object value = redisTemplate.opsForValue().get(BTS_PLAYBACK_KEY_PREFIX + historyKey);
- return value == null ? null : JSON.parseObject(value.toString(), BtsPlaybackState.class);
- }
-
- @Override
- public void saveBtsPlaybackState(String historyKey, BtsPlaybackState playback)
- {
- redisTemplate.opsForValue().set(BTS_PLAYBACK_KEY_PREFIX + historyKey, JSON.toJSONString(playback));
- }
-
- @Override
- public void deleteBtsPlaybackState(String historyKey)
- {
- redisTemplate.delete(BTS_PLAYBACK_KEY_PREFIX + historyKey);
- }
-
- @Override
- public void deleteBtsHistory(String historyKey)
- {
- String key = BTS_HISTORY_KEY_PREFIX + historyKey;
- redisTemplate.delete(key);
- redisTemplate.opsForSet().remove(BTS_HISTORY_KEYS_KEY, key);
- }
-
- @Override
- @SuppressWarnings({ "rawtypes", "unchecked" })
- public void saveBtsPlaybackAndAppendHistory(String historyKey, BtsPlaybackState playback,
- List<BtsHistoryRecord> records)
- {
- 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();
- String key = BTS_HISTORY_KEY_PREFIX + historyKey;
- if (!records.isEmpty())
- {
- typed.opsForList().rightPushAll(key, new ArrayList<>(records.stream()
- .map(JSON::toJSONString).toList()));
- typed.opsForList().trim(key, -MAX_HISTORY_RECORDS, -1);
- typed.opsForSet().add(BTS_HISTORY_KEYS_KEY, key);
- }
- typed.opsForValue().set(BTS_PLAYBACK_KEY_PREFIX + historyKey, JSON.toJSONString(playback));
- return typed.exec();
- }
- });
- if (results == null) throw new IllegalStateException("BTS simulator playback transaction failed in Redis");
- }
-
- @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");
+ redisTemplate.opsForValue().set(STATE_KEY, JSON.toJSONString(state));
}
@Override
@@ -307,21 +66,11 @@
@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);
- Set<Object> btsHistoryKeys = redisTemplate.opsForSet().members(BTS_HISTORY_KEYS_KEY);
- if (btsHistoryKeys != null && !btsHistoryKeys.isEmpty()) redisTemplate.delete(btsHistoryKeys);
- redisTemplate.delete(BTS_HISTORY_KEYS_KEY);
- // 播放状态按通道独立保存，重置演示时必须一并清理。
- Set<Object> playbackKeys = redisTemplate.keys(BTS_PLAYBACK_KEY_PREFIX + "*");
- if (playbackKeys != null && !playbackKeys.isEmpty()) redisTemplate.delete(playbackKeys);
redisTemplate.delete(RECENT_REQUESTS_KEY);
save(defaultState);
}
- private ParsedSimulatorState parseState(Object stored)
+ private SimulatorState restoreState(Object stored)
{
try
{
@@ -333,20 +82,13 @@
List<DeviceState> devices = new ArrayList<>();
requiredArray(root, "devices").forEach(item -> devices.add(deviceState(item)));
List<CollectorState> collectors = new ArrayList<>();
- 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));
- });
+ requiredArray(root, "collectors").forEach(item -> collectors.add(collectorState(jsonObject(item))));
List<ChannelTombstone> tombstones = new ArrayList<>();
requiredArray(root, "tombstones").forEach(item -> tombstones.add(channelTombstone(item)));
SimulatorState state = new SimulatorState(version, devices, collectors,
- requiredInteger(root, "nextDeviceId"), requiredInteger(root, "nextCollectorId"), tombstones);
- return new ParsedSimulatorState(state, embeddedHistory);
+ requiredInteger(root, "nextDeviceId"), requiredInteger(root, "nextCollectorId"), tombstones,
+ root.getBoolean("softwareAccessible"), root.getLongValue("observationSequence"));
+ return state;
}
catch (IllegalStateException exception)
{
@@ -406,46 +148,18 @@
requiredDecimal(object, "voltageMaxV"), requiredInteger(object, "sampleIntervalMs"));
}
- private List<CollectorMeasurementRecord> embeddedCollectorHistory(JSONObject object)
+ private CollectorState collectorState(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");
- // 仅含发现配置的旧状态首次恢复时，以演示默认值补齐采样状态。
- if (!object.containsKey("sampleIntervalMs"))
- return new CollectorState(collectorId, deviceCode, ip, mac, name, manufacturer, model,
- channelCount, online);
-
+ int count = requiredInteger(object, "channelCount");
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")));
+ JSONArray configured = object.getJSONArray("channels");
+ if (configured == null) channels.addAll(com.battery.device.simulator.DevelopmentCollectorSimulation.defaultChannels(count));
+ else configured.forEach(item -> channels.add(collectorChannelState(item)));
+ // 重启只恢复配置，不把上次响应或已废止的历史作为新观测。
+ return new CollectorState(requiredInteger(object, "collectorId"), object.getString("deviceCode"),
+ object.getString("ip"), object.getString("mac"), object.getString("name"),
+ object.getString("manufacturer"), object.getString("model"), count, object.getBooleanValue("online"),
+ com.battery.device.simulator.DevelopmentCollectorSimulation.DEFAULT_SAMPLE_INTERVAL_MS, channels, List.of());
}
private CollectorChannelState collectorChannelState(Object value)
@@ -453,7 +167,9 @@
JSONObject object = jsonObject(value);
List<CollectorMetricState> metrics = new ArrayList<>();
requiredArray(object, "metrics").forEach(item -> metrics.add(collectorMetricState(item)));
- return new CollectorChannelState(object.getString("channelCode"), metrics);
+ return new CollectorChannelState(object.getString("channelCode"), metrics,
+ object.getString("workStatus") == null ? DeviceWorkStatus.WORKING
+ : DeviceWorkStatus.valueOf(object.getString("workStatus")));
}
private CollectorMetricState collectorMetricState(Object value)
@@ -462,60 +178,6 @@
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"));
- }
-
- /** 从尚未拆分快照模型的旧状态中提取最新测量值。 */
- 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)
@@ -534,13 +196,6 @@
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 BigDecimal requiredDecimal(JSONObject object, String field)
{
BigDecimal value = object.getBigDecimal(field);
@@ -548,26 +203,5 @@
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); }
- }
}
--
Gitblit v1.7.1