推式来源的工步与循环改为从采样段推导

yeg

5 天以前 99935242ecccd94e76978e2b71cbc46b87d31dc6

 battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java

@@ -70,19 +70,28 @@
            TestWarningExceptionService warningExceptions, Clock clock)
    { this.mapper = mapper; this.tests = tests; this.adapter = adapter; this.warningExceptions = warningExceptions; this.clock = clock; }

    /** 轮询接收中测试，判定实时超限和来源离线异常，并按独立周期保存成功观测。 */
    /**
     * 接收调度入口：对全部接收中的测试判定实时超限与来源离线异常，并按各自节奏保存成功观测。
     *
     * <p>前置条件：测试处于“检测中 + 接收中”。预警判定按约 5 秒节奏，柜端逐行接收和工步、
     * 循环汇总按来源写库节奏限流；只有成功观测且可归属本测试的数据才写入事实表，不执行历史补采。</p>
     *
     * @throws org.springframework.dao.DataAccessException 任一写入失败时整个接收事务回滚，本轮不保存任何测试的数据
     */
    @Override @Transactional(rollbackFor = Exception.class)
    public void captureReceivedData()
    {
        for (TTest test : receivingTests()) {
            Instant now = clock.instant();
            Instant previous = lastRealtimeCheckAt.get(test.getTestId());
            // STEP 1: 按约 5 秒节奏判定当前来源离线事实与实时超限，预警不受采样入库周期限制。
            // 实时预警独立于持久化槽位，并与暂停、停止和删除共享测试事务锁。
            if (previous == null || !now.isBefore(previous.plusSeconds(5))) {
                warningExceptions.recordOfflineSources(test, Date.from(now));
                for (TTestBattery battery : batteries(test)) saveRealtime(test, battery);
                lastRealtimeCheckAt.put(test.getTestId(), now);
            }
            // STEP 2: 单独按来源写库节奏读取工步与循环汇总，避免每秒重查厂商库。
            for (TTestBattery battery : batteries(test)) saveDeviceAggregates(test, battery);
        }
    }
@@ -99,12 +108,21 @@
    }
    private List<TTestBattery> batteries(TTest test) { return test.getBatteries() == null ? List.of() : test.getBatteries(); }

    /**
     * 保存单个测试电池的一次实时观测：来源支持设备数据行时逐行落库，否则回落到轮询槽位。
     *
     * <p>前置条件：测试正在接收且绑定通道完整。柜端与采集器分别判断成功，失败来源的字段保持为空。</p>
     *
     * @param test   接收中的测试
     * @param battery 测试内的一个电池号及其通道绑定
     */
    private void saveRealtime(TTest test, TTestBattery battery)
    {
        if (adapter == null) return;
        TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) return;
        // 来源以设备数据行为接收单位时，柜端事实逐行落库（同秒多行都保留），槽位只留给采集器端。
        // STEP 1: 优先按来源水位逐行接收柜端事实（同秒多行都保留），槽位只留给采集器端。
        boolean rowBased = persistCabinetSamples(test, battery, binding);
        // STEP 2: 来源不按设备数据行接收时，落到轮询槽位，保持原来的组合观测语义。
        saveRealtimeSlot(test, battery, binding, !rowBased);
    }

@@ -112,6 +130,12 @@
     * 按来源水位逐行接收柜端事实。水位是"开始接收"或"继续接收"这一刻来源里的最新一行，
     * 因此水位之前的数据行即使仍在来源里也不进本测试，暂停期间的空档也不会被补回来。
     *
     * <p>前置条件：调用方持有该测试电池的接收游标。读取节奏按来源写库周期限流；
     * 实例缺失、来源不可达或读取失败时明确失败并留痕，不回退到其他实例。</p>
     *
     * @param test    接收中的测试
     * @param battery 测试内的一个电池号
     * @param binding 该电池号绑定的检测柜通道
     * @return 该来源是否按设备数据行接收；来源不支持或被拒绝时返回 false，调用方回落到轮询槽位
     */
    private boolean persistCabinetSamples(TTest test, TTestBattery battery, TTestChannelBinding binding)
@@ -121,7 +145,7 @@
        SampleCursor cursor = sampleCursors.computeIfAbsent(key, ignored -> new SampleCursor());
        synchronized (cursor)
        {
            // 开始接收与继续接收都会推进接收边界：边界变化即重新取水位，暂停期间的数据不补采。
            // STEP 1: 开始接收与继续接收都会推进接收边界，边界变化即重新取水位，暂停期间的数据不补采。
            if (!java.util.Objects.equals(cursor.boundary, boundary))
            {
                cursor.boundary = boundary; cursor.watermark = null; cursor.rowBased = null; cursor.sampledAt = null;
@@ -135,6 +159,7 @@
                return Boolean.TRUE.equals(cursor.rowBased);
            try
            {
                // STEP 2: 首次接触该边界时只记录水位，不把水位之前的历史行带入本测试。
                if (cursor.watermark == null)
                {
                    cursor.sampledAt = now;
@@ -149,6 +174,7 @@
                    cursor.rowBased = true;
                    return true;
                }
                // STEP 3: 逐行写入水位之后的设备事实，每行成功后推进水位。
                List<CabinetRealtime> rows = adapter.samplesSince(channel, cursor.watermark);
                cursor.sampledAt = now;
                for (CabinetRealtime row : rows)
@@ -173,15 +199,22 @@
     * 把一行设备事实写成一条实时记录；来源行身份保证重复接收幂等、同身份不同内容报冲突。
     * 归属边界由水位在设备时间里判定（见 {@link #persistCabinetSamples}），
     * 这里不用服务器取数时间反推设备时间。
     *
     * @param test    接收中的测试
     * @param battery 测试内的一个电池号
     * @param binding 该电池号绑定的检测柜通道
     * @param row     来源返回的一行柜端事实；设备时间或服务器接收时间缺失时丢弃不写
     */
    private void insertCabinetSample(TTest test, TTestBattery battery, TTestChannelBinding binding, CabinetRealtime row)
    {
        if (row.deviceTime() == null || row.serverReceivedAt() == null) return;
        // STEP 1: 先用本次成功观测判定实时超限，异常不依赖采样入库周期。
        Date observedAt = Date.from(row.serverReceivedAt());
        evaluateCabinet(test, binding, "VOLTAGE", BigDecimal.valueOf(row.voltageV()), "V", observedAt);
        evaluateCabinet(test, binding, "CURRENT", BigDecimal.valueOf(row.currentA()), "A", observedAt);
        evaluateCabinet(test, binding, "CAPACITY", BigDecimal.valueOf(row.capacityAh()), "Ah", observedAt);
        evaluateCabinet(test, binding, "ENERGY", BigDecimal.valueOf(row.energyWh()), "Wh", observedAt);
        // STEP 2: 按来源身份组装记录，槽位留空以允许同秒多行各自成立。
        String identity = sourceIdentity(binding);
        String recordId = row.sourceRecordId() == null || row.sourceRecordId().isBlank()
                ? "cabinet:" + row.sequenceId() : row.sourceRecordId();
@@ -205,7 +238,7 @@
        value.setExtensionMeasurementsJson(JSON.toJSONString(extension));
        value.setSourceIdentity(identity); value.setSourceRecordId(recordId);
        value.setContentSha256(hash(value));
        // 重复接收的同内容记录幂等忽略；同身份但内容不同时保留原事实，不覆盖。
        // STEP 3: 按来源行身份幂等；同内容重复接收忽略，同身份但内容不同时保留原事实并告警，不覆盖。
        String existing = mapper.selectRealtimeRecordHashBySourceRecord(battery.getTestBatteryId(), identity, recordId);
        if (existing != null)
        {
@@ -217,16 +250,27 @@
        mapper.insertRealtimeRecord(value);
    }

    /**
     * 向轮询槽位写入一条实时采样记录，柜端与采集器端的成功观测在同一行合并保存。
     *
     * <p>前置条件：调用方已判定该来源不按设备数据行接收。槽位按持久化周期对齐，同一槽位只写一行；
     * 柜端和采集器端分别判断成功，失败来源的身份、时间和指标保持为空。</p>
     *
     * @param test          接收中的测试
     * @param battery       测试内的一个电池号
     * @param binding       该电池号绑定的检测柜通道
     * @param includeCabinet 本次是否把柜端事实合并进该槽位
     */
    private void saveRealtimeSlot(TTest test, TTestBattery battery, TTestChannelBinding binding, boolean includeCabinet)
    {
        // 柜端和采集器端分别判断成功；同一槽位只写一行，失败来源的身份、时间和指标保持为空。
        // STEP 1: 柜端和采集器端分别判断成功，各取最近一次成功观测；同一槽位只写一行，失败来源的身份、时间和指标保持为空。
        Instant now = clock.instant(); long interval = persistenceInterval(binding); String key = test.getTestId() + ":" + battery.getTestBatteryId();
        CabinetRealtime cabinet = includeCabinet ? latestCabinet(binding) : null;
        if (cabinet != null && !accepted(test, cabinet.deviceTime(), cabinet.serverReceivedAt())) cabinet = null;
        CollectorRealtime collector = latestCollector(battery);
        Map<String, CollectorMeasurement> mapped = mappedCollectorMetrics(test, battery, collector);
        if (cabinet == null && (collector == null || !collector.online() || mapped.isEmpty())) return;
        // 当前成功测量先判定预警；槽位尚未到期或历史行被删除，都不能阻断实时告警。
        // STEP 2: 槽位尚未到期或历史行被删除都不能阻断实时告警，当前成功测量先判定预警。
        if (cabinet != null) {
            Date observedAt = Date.from(cabinet.serverReceivedAt());
            evaluateCabinet(test, binding, "VOLTAGE", BigDecimal.valueOf(cabinet.voltageV()), "V", observedAt);
@@ -238,11 +282,12 @@
            warningExceptions.evaluate(test, "COLLECTOR", battery.getCollectorId() + ":" + measurement.channelCode(),
                    measurement.metricCode(), measurement.value(), measurement.unit(), Date.from(measurement.serverReceivedAt()));
        }
        // STEP 3: 按持久化周期限流并对齐槽位；删除修订随数据库保留，重启后同一槽位也不能重新出现。
        Instant previous = lastSnapshotAt.get(key); if (previous != null && now.isBefore(previous.plusMillis(interval))) return;
        Instant collectorTime = mapped.values().stream().map(CollectorMeasurement::deviceTime).max(Instant::compareTo).orElse(null);
        Instant slot = Instant.ofEpochMilli((now.toEpochMilli() / interval) * interval);
        // 删除修订随数据库保留；重启后同一槽位也不能重新出现。
        if (mapper.wasRealtimeSlotDeleted(test.getTestId(), battery.getTestBatteryId(), slot.toEpochMilli())) return;
        // STEP 4: 组装组合观测行，只写成功来源的字段。
        TTestRealtimeRecord row = new TTestRealtimeRecord(); row.setTestId(test.getTestId()); row.setTestBatteryId(battery.getTestBatteryId()); row.setBindingId(binding.getBindingId());
        row.setAdapterCode(binding.getAdapterCode()); row.setCaptureSlotUtc(Date.from(slot)); row.setServerReceivedAtUtc(Date.from(now));
        if (cabinet != null) { row.setCabinetDevtype(binding.getDevtype()); row.setCabinetDevid(binding.getDevid()); row.setCabinetSubdevid(binding.getSubdevid()); row.setCabinetChlid(binding.getChlid()); row.setCabinetAdapterTimezoneId(cabinet.timezoneId()); }
@@ -261,6 +306,7 @@
        row.setExtensionMeasurementsJson(JSON.toJSONString(extension)); row.setSourceIdentity(sourceIdentity(binding));
        row.setSourceRecordId(sourceRecordId(cabinet, mapped));
        row.setContentSha256(hash(row));
        // STEP 5: 同一槽位内容不相同报冲突并保留原事实；插入被忽略时再核对一次并发写入。
        String old = mapper.selectRealtimeRecordContentHash(test.getTestId(), battery.getTestBatteryId(), row.getCaptureSlotUtc());
        if (old != null) { if (!old.equalsIgnoreCase(HexFormat.of().formatHex(row.getContentSha256()))) log.warn("实时采样内容冲突，拒绝覆盖 testId={}, testBatteryId={}, captureSlotUtc={}", test.getTestId(), battery.getTestBatteryId(), slot); return; }
        int inserted = mapper.insertRealtimeRecord(row);
@@ -278,18 +324,31 @@
        if (value != null) warningExceptions.evaluate(test, "CABINET", sourceIdentity(binding), metricCode, value, unit, occurredAt);
    }

    /**
     * 保存单个测试电池的工步与循环汇总事实，来源编号或明确的时间组合保证重复观测幂等。
     *
     * <p>前置条件：测试正在接收且绑定通道完整。只接受来源返回的统计记录，不从实时采样推算；
     * 已按授权物理删除的汇总不会因为重复观测被写回。</p>
     *
     * @param test    接收中的测试
     * @param battery 测试内的一个电池号及其通道绑定
     */
    private void saveDeviceAggregates(TTest test, TTestBattery battery)
    {
        // STEP 1: 按来源写库节奏限流，避免接收调度每秒重查厂商库。
        // 工步、循环只接受来源返回的统计记录，并以来源编号或明确的时间组合执行幂等。
        if (adapter == null) return; TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) return;
        String key = test.getTestId() + ":" + battery.getTestBatteryId();
        // 接收调度每秒跑一次，而来源按自己的写库节奏产出工步：不按来源节奏限流会每秒重查厂商库。
        // 汇总用来源声明的汇总间隔：它可能比实时行的间隔长（推式来源的落地表在本地）。
        Instant now = clock.instant(); Instant previous = lastAggregateAt.get(key);
        if (previous != null && now.isBefore(previous.plusMillis(persistenceInterval(binding)))) return;
        if (previous != null && now.isBefore(previous.plusMillis(aggregateInterval(binding)))) return;
        lastAggregateAt.put(key, now);
        // STEP 2: 读取来源的工步与循环汇总；来源不可达时本轮放弃，下一轮重试。
        ChannelIdentity channel = new ChannelIdentity(binding.getAdapterCode(), binding.getDevtype(), binding.getDevid(), binding.getSubdevid(), binding.getChlid());
        List<StepRecord> steps; List<CycleRecord> cycles; try { steps = adapter.stepRecords(channel); cycles = adapter.cycleRecords(channel); } catch (RuntimeException ex) { return; }
        String identity = sourceIdentity(binding);
        // STEP 3: 逐条写入可归属的实际工步，已删除的来源行跳过，内容不一致时保留原事实。
        for (StepRecord source : steps == null ? List.<StepRecord>of() : steps)
        {
            if (!accepted(test, source.endedAt(), source.serverReceivedAt())) continue;
@@ -305,6 +364,7 @@
                    log.warn("工步记录内容冲突，拒绝覆盖 testId={}, dedupeKey={}", test.getTestId(), row.getDedupeKey());
            }
        }
        // STEP 4: 逐条写入可归属的循环汇总，来源类别随记录保存以便区分推导值与设备事实。
        for (CycleRecord source : cycles == null ? List.<CycleRecord>of() : cycles)
        {
            if (!accepted(test, source.endedAt(), source.serverReceivedAt())) continue;
@@ -351,6 +411,7 @@
        return String.join(":", parts);
    }
    private long persistenceInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.persistenceIntervalMillis(channelIdentity(b))); } catch (RuntimeException ex) { return 60000L; } }
    private long aggregateInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.aggregateIntervalMillis(channelIdentity(b))); } catch (RuntimeException ex) { return 60000L; } }
    private static String sourceIdentity(TTestChannelBinding b) { return String.join(":", String.valueOf(b.getAdapterCode()), String.valueOf(b.getDevtype()), String.valueOf(b.getDevid()), String.valueOf(b.getSubdevid()), String.valueOf(b.getChlid())); }
    private static String timezone(TTestChannelBinding b) { return b.getAdapterTimezoneId() == null || b.getAdapterTimezoneId().isBlank() ? "Asia/Hong_Kong" : b.getAdapterTimezoneId(); }
    private static String blankToNull(String value) { return value == null || value.isBlank() ? null : value; }