From 370ef4f5e663fdfc1d19d2774dfc4ec1ed0af60e Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 29 八月 2026 14:57:48 +0800
Subject: [PATCH] 修复采集器历史状态持续序列化错误
---
ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java | 56 +++++++++++++++++++++++++++++++++++---------------------
1 files changed, 35 insertions(+), 21 deletions(-)
diff --git a/ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java b/ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java
index 369172a..c07d403 100644
--- a/ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java
+++ b/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)
{
--
Gitblit v1.7.1