From b78295e8da96c3e5e76befeef9582ce5273dbb74 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期二, 29 九月 2026 13:10:23 +0800
Subject: [PATCH] 按代码规范检查整改实体注释、Service Javadoc 与分层边界
---
battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java | 73 +++++++++++++++++++++++++++++++++---
1 files changed, 66 insertions(+), 7 deletions(-)
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java b/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
index a5a484e..5a77983 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
+++ b/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,8 +324,18 @@
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();
@@ -287,9 +343,11 @@
Instant now = clock.instant(); Instant previous = lastAggregateAt.get(key);
if (previous != null && now.isBefore(previous.plusMillis(persistenceInterval(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 +363,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;
--
Gitblit v1.7.1