修复接收中设备离线异常的记录与展示

yeg

2026-09-05 78f53edb9542a83d9579714cf4e1e8133fc89ee9

修复接收中设备离线异常的记录与展示

按测试及绑定来源聚合离线异常，暂停或停止后不再累加；复用异常处理流程并保留阈值异常。新增数据库迁移及状态、空值和去重回归验证。

 battery-business/src/main/java/com/battery/business/domain/TTestWarningException.java

@@ -4,11 +4,12 @@
import java.math.BigDecimal;
import java.util.Date;

/** 测试阈值预警异常事实。 */
/** 测试阈值超限或接收期间设备离线的异常事实。 */
public class TTestWarningException implements Serializable {
    private static final long serialVersionUID = 1L;
    private Long exceptionId;
    private Long testId;
    private String exceptionType = "THRESHOLD";
    private String testName;
    private String sourceType;
    private String sourceIdentity;
@@ -27,6 +28,7 @@
    private String handlingNote;
    public Long getExceptionId(){return exceptionId;} public void setExceptionId(Long v){exceptionId=v;}
    public Long getTestId(){return testId;} public void setTestId(Long v){testId=v;}
    public String getExceptionType(){return exceptionType;} public void setExceptionType(String v){exceptionType=v;}
    public String getTestName(){return testName;} public void setTestName(String v){testName=v;}
    public String getSourceType(){return sourceType;} public void setSourceType(String v){sourceType=v;}
    public String getSourceIdentity(){return sourceIdentity;} public void setSourceIdentity(String v){sourceIdentity=v;}

 battery-business/src/main/java/com/battery/business/domain/vo/OfflineTestSource.java

New file
@@ -0,0 +1,4 @@
package com.battery.business.domain.vo;

/** 测试绑定来源的持久化离线事实；检测柜使用完整协议通道身份。 */
public record OfflineTestSource(String sourceType, String sourceIdentity) {}

 battery-business/src/main/java/com/battery/business/mapper/TTestMapper.java

@@ -68,7 +68,12 @@
            @Param("metricCode") String metricCode, @Param("value") java.math.BigDecimal value,
            @Param("unit") String unit, @Param("ruleSnapshot") String ruleSnapshot,
            @Param("severity") String severity, @Param("occurredAt") Date occurredAt);
    /** 查询异常事实；不从设备状态或缺失数据推导异常。 */
    /** 查询绑定来源的离线事实，同一采集器的多通道绑定只返回一次。 */
    List<com.battery.business.domain.vo.OfflineTestSource> selectOfflineTestSources(Long testId);
    /** 以测试和来源身份聚合离线异常，不写入阈值或测量值。 */
    int upsertOfflineException(@Param("testId") Long testId, @Param("sourceType") String sourceType,
            @Param("sourceIdentity") String sourceIdentity, @Param("occurredAt") Date occurredAt);
    /** 查询已持久化的阈值及离线异常事实。 */
    List<TTestWarningException> selectWarningExceptions(@Param("testId") Long testId,
            @Param("status") String status);
    TTestWarningException selectWarningExceptionById(Long exceptionId);

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

@@ -46,7 +46,7 @@
        TTest test = tests.selectTestById(testId);
        List<CabinetCandidate> cabinets = candidates.snapshot(Instant.now()).cabinets();
        List<LiveSource> sources = new ArrayList<>();
        boolean receiving = "RECEIVING".equals(test.getReceptionStatus());
        boolean receiving = "RUNNING".equals(test.getTestStatus()) && "RECEIVING".equals(test.getReceptionStatus());
        for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
        {
            if (battery.getBinding() != null)
@@ -55,24 +55,53 @@
                sources.add(collectorSource(battery, receiving));
        }
        List<Map<String, Object>> exceptions = new ArrayList<>();
        if (receiving)
            exceptions.addAll(sources.stream().filter(source -> !"COLLECTOR".equals(source.sourceType())
                    && "OFFLINE".equals(source.deviceStatus()))
                    .map(source -> Map.<String, Object>of("sourceType", source.sourceType(), "sourceId", source.sourceId(),
                            "batteryNo", source.batteryNo(), "status", source.deviceStatus(), "message", "数据来源当前不可观测"))
                    .toList());
        java.util.Set<String> offlineSources = new java.util.HashSet<>();
        // 完成或暂停后只保留历史异常事实，实时控制台不再展示开放预警。
        if (receiving && mapper != null) {
            mapper.selectWarningExceptions(testId, "OPEN").forEach(item -> exceptions.add(Map.of(
                    "exceptionId", item.getExceptionId(), "sourceType", item.getSourceType(),
                    "metricCode", item.getMetricCode(), "value", item.getAbnormalValue(), "unit", item.getUnit(),
                    "status", item.getExceptionStatus(), "message", "测试预警异常")));
            mapper.selectWarningExceptions(testId, "OPEN").forEach(item -> {
                boolean offline = "DEVICE_OFFLINE".equals(item.getExceptionType());
                if (offline) offlineSources.add(item.getSourceType() + ":" + item.getSourceIdentity());
                // 离线异常没有测量值，必须允许空值，不能用 Map.of 构造响应。
                Map<String, Object> warning = new LinkedHashMap<>();
                warning.put("exceptionId", item.getExceptionId());
                warning.put("exceptionType", item.getExceptionType());
                warning.put("sourceType", item.getSourceType());
                warning.put("sourceIdentity", item.getSourceIdentity());
                warning.put("metricCode", item.getMetricCode());
                warning.put("value", item.getAbnormalValue());
                warning.put("unit", item.getUnit());
                warning.put("status", item.getExceptionStatus());
                warning.put("message", offline ? "接收期间设备离线，等待处理" : "测试预警异常");
                exceptions.add(warning);
            });
        }
        if (receiving) {
            for (LiveSource source : sources) {
                if (!"OFFLINE".equals(source.deviceStatus())) continue;
                String identity = offlineIdentity(test, source);
                if (!offlineSources.add(source.sourceType() + ":" + identity)) continue;
                exceptions.add(Map.of("exceptionType", "DEVICE_OFFLINE", "sourceType", source.sourceType(),
                        "sourceId", source.sourceId(), "sourceIdentity", identity,
                        "status", "OPEN", "message", "接收中数据来源离线，当前无法取得测量数据"));
            }
        }
        String testStatus = test.getTestStatus();
        TestPlanSnapshot testPlan = detectorTestPlan(test);
        return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now().toEpochMilli(), sources, exceptions, testPlan);
    }

    private String offlineIdentity(TTest test, LiveSource source)
    {
        if ("COLLECTOR".equals(source.sourceType())) return String.valueOf(source.sourceId());
        for (TTestBattery battery : test.getBatteries()) {
            var binding = battery.getBinding();
            if (binding != null && Objects.equals(binding.getChannelId(), source.sourceId()))
                return binding.getAdapterCode() + ":" + binding.getDevtype() + ":" + binding.getDevid()
                        + ":" + binding.getSubdevid() + ":" + binding.getChlid();
        }
        return String.valueOf(source.sourceId());
    }

    private TestPlanSnapshot detectorTestPlan(TTest test)
    {
        if (adapter == null) return null;

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

@@ -46,6 +46,7 @@
    private final TestWarningExceptionService warningExceptions;
    private final Clock clock;
    private final Map<String, Instant> lastSnapshotAt = new java.util.concurrent.ConcurrentHashMap<>();
    private final Map<Long, Instant> lastOfflineCheckAt = new java.util.concurrent.ConcurrentHashMap<>();

    @Autowired
    public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, @Nullable TestDataObservationAdapter adapter,
@@ -63,11 +64,20 @@
            TestWarningExceptionService warningExceptions, Clock clock)
    { this.mapper = mapper; this.tests = tests; this.adapter = adapter; this.warningExceptions = warningExceptions; this.clock = clock; }

    /** 轮询接收中测试，只保存本次成功观测的实时槽位及设备直接返回汇总。 */
    /** 轮询接收中测试，记录绑定来源离线异常，并保存成功观测的实时槽位及设备汇总。 */
    @Override @Transactional(rollbackFor = Exception.class)
    public void captureReceivedData()
    {
        for (TTest test : receivingTests()) for (TTestBattery battery : batteries(test)) { saveRealtime(test, battery); saveDeviceAggregates(test, battery); }
        for (TTest test : receivingTests()) {
            Instant now = clock.instant();
            Instant previous = lastOfflineCheckAt.get(test.getTestId());
            // 离线检查独立于测量采样槽位，并与暂停、停止和删除共享测试事务锁。
            if (previous == null || !now.isBefore(previous.plusSeconds(5))) {
                warningExceptions.recordOfflineSources(test, Date.from(now));
                lastOfflineCheckAt.put(test.getTestId(), now);
            }
            for (TTestBattery battery : batteries(test)) { saveRealtime(test, battery); saveDeviceAggregates(test, battery); }
        }
    }

    private List<TTest> receivingTests()

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

@@ -10,11 +10,19 @@
import com.battery.business.domain.TTestWarningThreshold;
import com.battery.business.mapper.TTestMapper;

/** 对成功接收的测量值执行测试阈值判断；设备状态和缺失数据不经过此服务。 */
/** 分别记录测量超限与接收中的离线事实；缺失测量不参与阈值判断。 */
@Service
public class TestWarningExceptionService {
    private final TTestMapper mapper;
    public TestWarningExceptionService(TTestMapper mapper) { this.mapper = mapper; }
    /** 调用方须持有测试事务锁；仅在接收中记录当前绑定来源的离线事实。 */
    public void recordOfflineSources(TTest test, Date occurredAt) {
        if (test == null || !"RUNNING".equals(test.getTestStatus()) || !"RECEIVING".equals(test.getReceptionStatus())) return;
        for (var source : mapper.selectOfflineTestSources(test.getTestId())) {
            mapper.upsertOfflineException(test.getTestId(), source.sourceType(), source.sourceIdentity(), occurredAt);
        }
    }

    /** 仅对成功接收的非空测量值判定阈值，并聚合更新开放异常及规则快照。 */
    public void evaluate(TTest test, String sourceType, String sourceIdentity, String metricCode, BigDecimal value, String unit, Date occurredAt) {
        if (value == null || test == null || test.getThresholds() == null) return;

 battery-business/src/main/resources/mapper/battery/TTestMapper.xml

@@ -97,6 +97,7 @@
    </resultMap>
    <resultMap id="TTestWarningExceptionResultMap" type="com.battery.business.domain.TTestWarningException">
        <id property="exceptionId" column="exception_id"/><result property="testId" column="test_id"/>
        <result property="exceptionType" column="exception_type"/>
        <result property="testName" column="test_name"/><result property="sourceType" column="source_type"/>
        <result property="sourceIdentity" column="source_identity"/><result property="metricCode" column="metric_code"/>
        <result property="abnormalValue" column="abnormal_value"/><result property="unit" column="unit"/>
@@ -313,6 +314,29 @@
            last_occurred_at_utc = values(last_occurred_at_utc), occurrence_count = occurrence_count + 1,
            source_identity = values(source_identity), severity = values(severity), rule_snapshot = values(rule_snapshot)
    </insert>
    <select id="selectOfflineTestSources" parameterType="Long" resultType="com.battery.business.domain.vo.OfflineTestSource">
        select distinct 'CABINET_CHANNEL' as source_type,
               concat(c.adapter_code, ':', c.devtype, ':', c.devid, ':', c.subdevid, ':', c.chlid) as source_identity
        from t_test_channel_binding b
        join t_detection_channel c on c.channel_id = b.channel_id
        join t_detection_cabinet cabinet on cabinet.cabinet_id = c.cabinet_id
        where b.test_id = #{testId} and (c.online_status = 'OFFLINE' or cabinet.online_status = 'OFFLINE')
        union
        select 'COLLECTOR' as source_type, cast(c.collector_id as char) as source_identity
        from t_test_collector_channel b
        join t_collector c on c.collector_id = b.collector_id
        where b.test_id = #{testId} and c.online_status = 'OFFLINE'
    </select>
    <insert id="upsertOfflineException">
        insert into t_test_warning_exception
        (test_id, exception_type, threshold_id, source_type, source_identity, metric_code,
         abnormal_value, unit, rule_snapshot, severity, exception_status,
         first_occurred_at_utc, last_occurred_at_utc, occurrence_count)
        values (#{testId}, 'DEVICE_OFFLINE', null, #{sourceType}, #{sourceIdentity}, 'DEVICE_ONLINE',
                null, '', '{"expectedStatus":"ONLINE"}', 'WARNING', 'OPEN', #{occurredAt}, #{occurredAt}, 1)
        on duplicate key update last_occurred_at_utc = values(last_occurred_at_utc),
            occurrence_count = occurrence_count + 1
    </insert>
    <select id="selectWarningExceptions" resultMap="TTestWarningExceptionResultMap">
        select e.*, t.test_name, u.nick_name handled_by_name
        from t_test_warning_exception e join t_test t on t.test_id=e.test_id

 ruoyi-admin/src/main/resources/db/migration-final/V7__offline_test_exceptions.sql

New file
@@ -0,0 +1,15 @@
-- 离线异常不关联阈值；已有阈值异常保持原聚合规则。
ALTER TABLE t_test_warning_exception
    ADD COLUMN exception_type VARCHAR(32) NOT NULL DEFAULT 'THRESHOLD' AFTER test_id,
    MODIFY COLUMN threshold_id BIGINT NULL,
    DROP INDEX uk_warning_exception_open,
    DROP COLUMN open_dedupe_key;

ALTER TABLE t_test_warning_exception
    ADD COLUMN open_dedupe_key VARCHAR(512) GENERATED ALWAYS AS
        (CASE WHEN exception_status = 'OPEN' THEN
            CASE WHEN exception_type = 'DEVICE_OFFLINE'
                 THEN CONCAT('offline:', test_id, ':', source_type, ':', source_identity)
                 ELSE CONCAT(test_id, ':', threshold_id, ':', metric_code) END
            ELSE NULL END) STORED,
    ADD UNIQUE KEY uk_warning_exception_open (open_dedupe_key);

 ruoyi-admin/src/test/java/com/battery/business/service/impl/TTestLiveServiceImplTest.java

@@ -38,6 +38,48 @@
class TTestLiveServiceImplTest
{
    @Test
    void persistedOfflineWarningWithoutMeasurementReplacesLiveHintAndSurvivesRefresh()
    {
        TTest test = test("RECEIVING");
        var mapper = mock(TTestMapper.class);
        TTestWarningException warning = new TTestWarningException();
        warning.setExceptionId(21L);
        warning.setExceptionType("DEVICE_OFFLINE");
        warning.setSourceType("CABINET_CHANNEL");
        warning.setSourceIdentity("dev-sim:1:1:0:1");
        warning.setMetricCode("DEVICE_ONLINE");
        warning.setExceptionStatus("OPEN");
        when(mapper.selectWarningExceptions(10L, "OPEN")).thenReturn(List.of(warning));
        var service = liveService(test, Instant.now(), false, DeviceWorkStatus.WORKING, mapper);

        for (int attempt = 0; attempt < 2; attempt++) {
            var alarms = service.current(10L).exceptions();
            assertEquals(1, alarms.size());
            assertEquals(21L, alarms.get(0).get("exceptionId"));
            assertEquals("DEVICE_OFFLINE", alarms.get(0).get("exceptionType"));
        }
        test.setReceptionStatus("PAUSED");
        assertTrue(service.current(10L).exceptions().isEmpty());
        test.setReceptionStatus("ENDED");
        assertTrue(service.current(10L).exceptions().isEmpty());
    }

    @Test
    void receivingOfflineCabinetAndCollectorAreVisibleAsOpenAlarms()
    {
        TTest test = test("RECEIVING");
        test.getBatteries().get(0).setCollectorId(7L);
        var service = liveService(test, Instant.now(), false);

        var alarms = service.current(10L).exceptions();

        assertEquals(2, alarms.size());
        assertTrue(alarms.stream().allMatch(item -> "OPEN".equals(item.get("status"))));
        assertTrue(alarms.stream().anyMatch(item -> "COLLECTOR".equals(item.get("sourceType"))));
        assertTrue(alarms.stream().anyMatch(item -> "CABINET_CHANNEL".equals(item.get("sourceType"))));
    }

    @Test
    void nonReceivingConsoleReadsCollectorStatusWithoutRequestingMeasurements() throws Exception
    {
        try (var simulator = new com.battery.device.simulator.BtsDevelopmentSimulator(0))

 ruoyi-admin/src/test/java/com/battery/business/service/impl/TestDataReceptionServiceImplTest.java

@@ -48,6 +48,23 @@
    private static final Clock CLOCK = Clock.fixed(CAPTURED_AT, ZoneOffset.UTC);

    @Test
    void recordsOfflineSourcesInBackgroundEvenWithoutSuccessfulSamples() {
        Fixture fixture = fixture();
        when(fixture.mapper.selectOfflineTestSources(1L)).thenReturn(List.of(
                new com.battery.business.domain.vo.OfflineTestSource("COLLECTOR", "9")));
        var service = fixture.service();

        service.captureReceivedData();
        service.captureReceivedData();

        verify(fixture.mapper).upsertOfflineException(1L, "COLLECTOR", "9", Date.from(CAPTURED_AT));
        verify(fixture.mapper, never()).insertRealtimeRecord(any());
        when(fixture.mapper.selectTestByIdForUpdate(1L)).thenReturn(null);
        service.captureReceivedData();
        verify(fixture.mapper, times(1)).selectOfflineTestSources(1L);
    }

    @Test
    void savesOneCombinedRealtimeRowForOneBatteryPollingSlot()
    {
        Fixture fixture = fixture();

 ruoyi-admin/src/test/java/com/battery/business/service/impl/TestWarningExceptionServiceTest.java

@@ -13,9 +13,33 @@
import com.battery.business.domain.TTestWarningThreshold;
import com.battery.business.mapper.TTestMapper;

/** 测试预警仅由成功接收测量值触发，并通过持久层原子聚合。 */
/** 阈值异常与接收中的离线异常分别判定，并通过持久层原子聚合。 */
class TestWarningExceptionServiceTest {
    @Test
    void recordsOfflineSourcesOnlyWhileReceivingWithoutInventingMeasurements() {
        TTestMapper mapper = mock(TTestMapper.class);
        var service = new TestWarningExceptionService(mapper);
        TTest test = new TTest();
        test.setTestId(1L);
        test.setTestStatus("RUNNING");
        test.setReceptionStatus("RECEIVING");
        Date now = new Date();
        when(mapper.selectOfflineTestSources(1L)).thenReturn(List.of(
                new com.battery.business.domain.vo.OfflineTestSource("CABINET_CHANNEL", "sim:22:1:1:1"),
                new com.battery.business.domain.vo.OfflineTestSource("COLLECTOR", "9")));

        service.recordOfflineSources(test, now);
        verify(mapper).upsertOfflineException(1L, "CABINET_CHANNEL", "sim:22:1:1:1", now);
        verify(mapper).upsertOfflineException(1L, "COLLECTOR", "9", now);
        clearInvocations(mapper);
        for (String state : List.of("PAUSED", "ENDED", "NOT_STARTED")) {
            test.setReceptionStatus(state);
            service.recordOfflineSources(test, now);
        }
        verifyNoInteractions(mapper);
    }

    @Test
    void evaluatesOnlyOutOfRangeValues() {
        TTestMapper mapper = mock(TTestMapper.class);
        TestWarningExceptionService service = new TestWarningExceptionService(mapper);