实现采集器独立实时与历史采样模拟

yeg

2026-08-29 7ab0e426264707fb85ee38d0a9a73caf41899635

 ruoyi-device/src/main/java/com/ruoyi/device/bts/BtsDevelopmentSimulator.java

@@ -12,11 +12,18 @@
import com.ruoyi.device.simulator.BtsSimulatorStateStore;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.ChannelTombstone;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorChannelState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorHistoryFaults;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMetricState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.DeviceState;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.RecentRequest;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.SimulatorState;
import com.ruoyi.device.simulator.InMemoryBtsSimulatorStateStore;
import com.ruoyi.device.simulator.DevelopmentCollectorSimulation;
import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorHistoryPage;
import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;

/** 提供由开发后端托管的临时只读 BTS 模拟端点。 */
public final class BtsDevelopmentSimulator implements AutoCloseable
@@ -138,8 +145,7 @@
        CollectorState current = requireCollector(collectorId);
        List<CollectorState> collectors = new ArrayList<>(state.collectors());
        collectors.replaceAll(item -> item.collectorId() == collectorId
                ? new CollectorState(collectorId, current.deviceCode(), ip.trim(), current.mac(), current.name(),
                        current.manufacturer(), current.model(), channelCount, online)
                ? DevelopmentCollectorSimulation.resize(current, ip, channelCount, online, Instant.now())
                : item);
        persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
                state.nextCollectorId(), state.tombstones()));
@@ -155,6 +161,45 @@
                state.nextCollectorId(), state.tombstones()));
    }

    /** 配置采集器独立采样频率和历史下载故障场景。 */
    public synchronized void configureCollectorSampling(int collectorId, int sampleIntervalMs,
            CollectorHistoryFaults historyFaults)
    {
        CollectorState current = requireCollector(collectorId);
        replaceCollector(DevelopmentCollectorSimulation.configureSampling(current, sampleIntervalMs, historyFaults));
    }

    /** 配置一个稳定 S 通道的固定或动态演示指标。 */
    public synchronized void configureCollectorChannel(int collectorId, String channelCode,
            List<CollectorMetricState> metrics)
    {
        CollectorState current = requireCollector(collectorId);
        replaceCollector(DevelopmentCollectorSimulation.configureChannel(current, channelCode, metrics));
    }

    /** 按每台采集器自己的频率推进实时和历史采样。 */
    public synchronized void advanceCollectors(Instant now)
    {
        List<CollectorState> collectors = state.collectors().stream()
                .map(collector -> DevelopmentCollectorSimulation.advance(collector, now))
                .toList();
        if (!collectors.equals(state.collectors()))
            persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
                    state.nextCollectorId(), state.tombstones()));
    }

    /** 返回采集器最新实时快照；快照与分页历史保持不同读取路径。 */
    public synchronized CollectorRealtimeSnapshot collectorRealtime(int collectorId)
    {
        return DevelopmentCollectorSimulation.realtime(requireCollector(collectorId));
    }

    /** 分页下载采集器模拟历史，并应用已配置的重复、冲突、乱序或缺失场景。 */
    public synchronized CollectorHistoryPage collectorHistory(int collectorId, long offset, int limit)
    {
        return DevelopmentCollectorSimulation.history(requireCollector(collectorId), offset, limit);
    }

    /** 返回控制页需要的运行状态快照。 */
    public synchronized StateSnapshot snapshot()
    {
@@ -168,7 +213,9 @@
        List<CollectorSnapshot> collectorSnapshots = state.collectors().stream()
                .map(collector -> new CollectorSnapshot(collector.collectorId(), collector.deviceCode(),
                        collector.ip(), collector.mac(), collector.name(), collector.manufacturer(),
                        collector.model(), collector.channelCount(), collector.online()))
                        collector.model(), collector.channelCount(), collector.online(), collector.sampleIntervalMs(),
                        collector.channels(), collector.historyFaults(), collector.history().size(),
                        DevelopmentCollectorSimulation.realtime(collector).measurements()))
                .toList();
        return new StateSnapshot(running, HOST + ":" + port(), deviceSnapshots, collectorSnapshots,
                stateStore.loadRecentRequests());
@@ -202,6 +249,14 @@
        return new CollectorState(collectorId, "dev-sim-collector-" + collectorId,
                ip.trim(), simulatorCollectorMac(collectorId), "dev-sim 采集器 " + collectorId,
                "开发模拟器", "Virtual Collector", channelCount, true);
    }

    private void replaceCollector(CollectorState replacement)
    {
        List<CollectorState> collectors = new ArrayList<>(state.collectors());
        collectors.replaceAll(item -> item.collectorId() == replacement.collectorId() ? replacement : item);
        persist(new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(),
                state.nextCollectorId(), state.tombstones()));
    }

    private String respond(String request)
@@ -442,6 +497,7 @@
                    || !isIpv4(collector.ip()) || !ips.add(collector.ip())
                    || collector.channelCount() < 1 || collector.channelCount() > MAX_CHANNEL_COUNT)
                throw new IllegalStateException("Invalid BTS simulator collector state in Redis");
            DevelopmentCollectorSimulation.validate(collector);
            maximumCollectorId = Math.max(maximumCollectorId, collector.collectorId());
        }
        java.util.Set<String> tombstoneIds = new java.util.HashSet<>();
@@ -471,6 +527,15 @@
    public record ChannelSnapshot(int channelId, boolean online, DeviceWorkStatus workStatus) { }

    public record CollectorSnapshot(int collectorId, String deviceCode, String ip, String mac, String name,
            String manufacturer, String model, int channelCount, boolean online) { }
            String manufacturer, String model, int channelCount, boolean online, int sampleIntervalMs,
            List<CollectorChannelState> channels, CollectorHistoryFaults historyFaults, int historyRecordCount,
            List<CollectorMeasurementRecord> latestMeasurements)
    {
        public CollectorSnapshot
        {
            channels = List.copyOf(channels);
            latestMeasurements = List.copyOf(latestMeasurements);
        }
    }

}