优化采集器长时历史的有界读取

yeg

2026-08-29 8761644af158c5e7f05e999e9834b6496ce414e7

 ruoyi-admin/src/main/java/com/ruoyi/config/RedisBtsSimulatorStateStore.java

@@ -25,7 +25,6 @@
    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();
@@ -99,25 +98,56 @@
    }

    @Override
    public List<CollectorMeasurementRecord> loadCollectorHistory(String collectorCode)
    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);
        List<CollectorMeasurementRecord> records = new ArrayList<>();
        if (storedSize != null)
        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)
        {
            for (long start = 0; start < storedSize; start += HISTORY_READ_BATCH_SIZE)
            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)
            {
                long end = Math.min(start + HISTORY_READ_BATCH_SIZE, storedSize) - 1;
                List<Object> stored = redisTemplate.opsForList().range(key, start, end);
                List<Object> stored = redisTemplate.opsForList().range(key, lower, redisUpper - 1);
                if (stored != null)
                    stored.forEach(item -> records.add(collectorMeasurementRecord(item)));
                    for (int index = stored.size() - 1; index >= 0; index--)
                        records.add(collectorMeasurementRecord(stored.get(index)));
            }
        }
        pendingEmbeddedHistory.stream()
                .filter(append -> append.collectorCode().equals(collectorCode))
                .forEach(append -> records.addAll(append.records()));
        return List.copyOf(records);
        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