From 8761644af158c5e7f05e999e9834b6496ce414e7 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 29 八月 2026 16:04:31 +0800
Subject: [PATCH] 优化采集器长时历史的有界读取
---
ruoyi-device/src/main/java/com/ruoyi/device/simulator/DevelopmentCollectorSimulation.java | 127 +++++++++++++++++++++++++++++-------------
1 files changed, 88 insertions(+), 39 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 c07d403..c0fb1fd 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
@@ -8,7 +8,6 @@
import java.time.format.DateTimeParseException;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
-import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -26,7 +25,7 @@
public static final int DEFAULT_SAMPLE_INTERVAL_MS = 1000;
public static final int MIN_SAMPLE_INTERVAL_MS = 100;
public static final int MAX_SAMPLE_INTERVAL_MS = 60000;
- public static final int MAX_HISTORY_PAGE_SIZE = 1000;
+ public static final int MAX_HISTORY_PAGE_SIZE = BtsSimulatorStateStore.MAX_HISTORY_BATCH_SIZE;
private static final int MAX_SAMPLES_PER_ADVANCE = 1000;
private static final ZoneId DEVICE_ZONE = ZoneId.of("Asia/Hong_Kong");
private static final DateTimeFormatter DEVICE_TIME_FORMATTER =
@@ -145,22 +144,26 @@
}
/** 对只追加原始模拟历史应用可重复的下载故障视图，再按偏移量分页。 */
- public static CollectorHistoryPage history(CollectorState current, List<CollectorMeasurementRecord> rawHistory,
+ public static CollectorHistoryPage history(CollectorState current, BtsSimulatorStateStore stateStore,
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, 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(),
- download.size(), download.subList(from, to));
+ CollectorHistoryFaults faults = current.historyFaults();
+ if (faults.missingSequences().isEmpty() && faults.duplicateSequence() == null
+ && faults.conflictingSequence() == null && !faults.outOfOrder())
+ {
+ BtsSimulatorStateStore.CollectorHistoryBatch batch = stateStore.loadCollectorHistoryPage(
+ current.deviceCode(), offset, limit, false);
+ return new CollectorHistoryPage(current.deviceCode(), offset, batch.nextOffset(), batch.hasMore(),
+ batch.total(), batch.records());
+ }
+ return faultedHistoryPage(current, stateStore, offset, limit);
}
/** 校验 Redis 恢复的采集器采样状态，拒绝损坏状态被静默替换。 */
- public static void validate(CollectorState state, List<CollectorMeasurementRecord> rawHistory)
+ public static void validate(CollectorState state, BtsSimulatorStateStore stateStore)
{
validateSampleInterval(state.sampleIntervalMs());
if (state.nextSequence() < 1 || state.channels().size() != state.channelCount())
@@ -173,7 +176,7 @@
validateMetrics(channel.metrics());
}
validateLatestMeasurements(state);
- validateHistory(state, rawHistory);
+ validateHistory(state, stateStore);
validateFaults(state.historyFaults());
}
@@ -193,25 +196,49 @@
}
}
- private static void validateHistory(CollectorState state, List<CollectorMeasurementRecord> rawHistory)
+ private static void validateHistory(CollectorState state, BtsSimulatorStateStore stateStore)
{
long maximumSequence = 0;
long previousSequence = 0;
+ long recordCount = 0;
+ long offset = 0;
Set<String> identities = new java.util.HashSet<>();
- for (CollectorMeasurementRecord record : rawHistory)
+ while (true)
{
- String identity = record.sequenceId() + "\u0000" + record.channelCode() + "\u0000" + record.metricCode();
- if (!validMeasurement(state, record.adapterCode(), record.collectorCode(), record.channelCode(),
- record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(), record.serverReceivedAt(),
- record.metricCode(), record.value(), record.unit()) || !identities.add(identity)
- || record.sequenceId() < previousSequence)
- throw new IllegalStateException("Invalid collector history state in Redis");
- maximumSequence = Math.max(maximumSequence, record.sequenceId());
- previousSequence = record.sequenceId();
+ BtsSimulatorStateStore.CollectorHistoryBatch batch = stateStore.loadCollectorHistoryPage(
+ state.deviceCode(), offset, MAX_HISTORY_PAGE_SIZE, false);
+ if (batch.offset() != offset || batch.records().size() > MAX_HISTORY_PAGE_SIZE)
+ throw new IllegalStateException("Invalid collector history page in Redis");
+ for (CollectorMeasurementRecord record : batch.records())
+ {
+ if (record.sequenceId() < previousSequence)
+ throw new IllegalStateException("Invalid collector history state in Redis");
+ if (record.sequenceId() != previousSequence)
+ {
+ identities.clear();
+ previousSequence = record.sequenceId();
+ }
+ String identity = record.channelCode() + "\u0000" + record.metricCode();
+ if (!validMeasurement(state, record.adapterCode(), record.collectorCode(), record.channelCode(),
+ record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(), record.serverReceivedAt(),
+ record.metricCode(), record.value(), record.unit()) || !identities.add(identity))
+ throw new IllegalStateException("Invalid collector history state in Redis");
+ maximumSequence = Math.max(maximumSequence, record.sequenceId());
+ recordCount++;
+ }
+ if (!batch.hasMore())
+ {
+ if (batch.total() != recordCount)
+ throw new IllegalStateException("Invalid collector history count in Redis");
+ break;
+ }
+ if (batch.nextOffset() <= offset)
+ throw new IllegalStateException("Invalid collector history page in Redis");
+ offset = batch.nextOffset();
}
if (state.nextSequence() <= maximumSequence)
throw new IllegalStateException("Invalid collector history sequence in Redis");
- if (state.historyRecordCount() != rawHistory.size())
+ if (state.historyRecordCount() != recordCount)
throw new IllegalStateException("Invalid collector history count in Redis");
}
@@ -272,28 +299,50 @@
return channelCode + "\u0000" + metricCode;
}
- private static List<CollectorMeasurementRecord> faultedHistory(CollectorState current,
- List<CollectorMeasurementRecord> rawHistory)
+ private static CollectorHistoryPage faultedHistoryPage(CollectorState current,
+ BtsSimulatorStateStore stateStore, long offset, int limit)
{
+ // 故障视图需返回精确 total，因此逐批扫描原始历史；正常下载会直接读取目标页。
CollectorHistoryFaults faults = current.historyFaults();
Set<Long> missing = Set.copyOf(faults.missingSequences());
- List<CollectorMeasurementRecord> records = new ArrayList<>();
- for (CollectorMeasurementRecord record : rawHistory)
+ List<CollectorMeasurementRecord> page = new ArrayList<>(limit);
+ long rawOffset = 0;
+ long total = 0;
+ while (true)
{
- if (missing.contains(record.sequenceId())) continue;
- records.add(record);
- if (Long.valueOf(record.sequenceId()).equals(faults.duplicateSequence())) records.add(record);
- if (Long.valueOf(record.sequenceId()).equals(faults.conflictingSequence()))
- records.add(new CollectorMeasurementRecord(record.adapterCode(), record.collectorCode(),
- record.channelCode(), record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(),
- record.serverReceivedAt(), record.metricCode(),
- record.value().add(new BigDecimal("0.001")), record.unit()));
+ BtsSimulatorStateStore.CollectorHistoryBatch batch = stateStore.loadCollectorHistoryPage(
+ current.deviceCode(), rawOffset, MAX_HISTORY_PAGE_SIZE, faults.outOfOrder());
+ for (CollectorMeasurementRecord record : batch.records())
+ {
+ if (missing.contains(record.sequenceId())) continue;
+ total = appendFaultedRecord(page, record, total, offset, limit);
+ if (Long.valueOf(record.sequenceId()).equals(faults.duplicateSequence()))
+ total = appendFaultedRecord(page, record, total, offset, limit);
+ if (Long.valueOf(record.sequenceId()).equals(faults.conflictingSequence()))
+ total = appendFaultedRecord(page, conflictingRecord(record), total, offset, limit);
+ }
+ if (!batch.hasMore()) break;
+ if (batch.nextOffset() <= rawOffset)
+ throw new IllegalStateException("Invalid collector history page in Redis");
+ rawOffset = batch.nextOffset();
}
- if (faults.outOfOrder())
- records.sort(Comparator.comparingLong(CollectorMeasurementRecord::sequenceId).reversed()
- .thenComparing(CollectorMeasurementRecord::channelCode)
- .thenComparing(CollectorMeasurementRecord::metricCode));
- return records;
+ long from = Math.min(offset, total);
+ long nextOffset = from + page.size();
+ return new CollectorHistoryPage(current.deviceCode(), offset, nextOffset, nextOffset < total, total, page);
+ }
+
+ private static long appendFaultedRecord(List<CollectorMeasurementRecord> page,
+ CollectorMeasurementRecord record, long index, long offset, int limit)
+ {
+ if (index >= offset && page.size() < limit) page.add(record);
+ return index + 1;
+ }
+
+ private static CollectorMeasurementRecord conflictingRecord(CollectorMeasurementRecord record)
+ {
+ return new CollectorMeasurementRecord(record.adapterCode(), record.collectorCode(), record.channelCode(),
+ record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(), record.serverReceivedAt(),
+ record.metricCode(), record.value().add(new BigDecimal("0.001")), record.unit());
}
private static CollectorState copy(CollectorState current, String ip, int channelCount, boolean online,
--
Gitblit v1.7.1