package com.ruoyi.energymeter.service.impl;
|
|
import com.ruoyi.common.exception.ServiceException;
|
import com.ruoyi.common.utils.StringUtils;
|
import com.ruoyi.energymeter.config.EnergyMeterProperties;
|
import com.ruoyi.energymeter.constant.EnergyMeterConstants;
|
import com.ruoyi.energymeter.dto.MeterReadingPoint;
|
import com.ruoyi.energymeter.dto.MeterUsagePoint;
|
import com.ruoyi.energymeter.dto.TaskStatusCount;
|
import com.ruoyi.energymeter.mapper.EnergyMeterDeviceMapper;
|
import com.ruoyi.energymeter.mapper.EnergyMeterMeasurementMapper;
|
import com.ruoyi.energymeter.mapper.EnergyMeterSyncCheckpointMapper;
|
import com.ruoyi.energymeter.mapper.EnergyMeterSyncRunMapper;
|
import com.ruoyi.energymeter.mapper.EnergyMeterSyncTaskMapper;
|
import com.ruoyi.energymeter.pojo.EnergyMeterDevice;
|
import com.ruoyi.energymeter.pojo.EnergyMeterMeasurement;
|
import com.ruoyi.energymeter.pojo.EnergyMeterSyncCheckpoint;
|
import com.ruoyi.energymeter.pojo.EnergyMeterSyncRun;
|
import com.ruoyi.energymeter.pojo.EnergyMeterSyncTask;
|
import com.ruoyi.energymeter.provider.MeterProviderClient;
|
import com.ruoyi.energymeter.provider.MeterProviderException;
|
import com.ruoyi.energymeter.service.EnergyMeterSyncService;
|
import com.ruoyi.energymeter.util.EnergyMeterSupport;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.beans.factory.annotation.Qualifier;
|
import org.springframework.dao.DuplicateKeyException;
|
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
import org.springframework.stereotype.Service;
|
import org.springframework.transaction.PlatformTransactionManager;
|
import org.springframework.transaction.support.TransactionSynchronization;
|
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
import org.springframework.transaction.support.TransactionTemplate;
|
|
import java.math.BigDecimal;
|
import java.time.LocalDate;
|
import java.time.LocalDateTime;
|
import java.time.format.DateTimeFormatter;
|
import java.time.temporal.ChronoUnit;
|
import java.util.ArrayList;
|
import java.util.Comparator;
|
import java.util.HashMap;
|
import java.util.LinkedHashMap;
|
import java.util.LinkedHashSet;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.Set;
|
import java.util.UUID;
|
import java.util.concurrent.CompletableFuture;
|
import java.util.stream.Collectors;
|
|
/**
|
* 同步编排：批次创建、任务执行、幂等落库、进度推进与失败重试。
|
*
|
* <p>外部 HTTP 调用不放在数据库事务中；写入采集表、更新任务和进度在短事务内提交。</p>
|
*/
|
@Slf4j
|
@Service
|
public class EnergyMeterSyncServiceImpl implements EnergyMeterSyncService {
|
|
private static final int UPSERT_CHUNK = 500;
|
|
private final EnergyMeterDeviceMapper deviceMapper;
|
private final EnergyMeterMeasurementMapper measurementMapper;
|
private final EnergyMeterSyncRunMapper runMapper;
|
private final EnergyMeterSyncTaskMapper taskMapper;
|
private final EnergyMeterSyncCheckpointMapper checkpointMapper;
|
private final MeterProviderClient provider;
|
private final EnergyMeterProperties properties;
|
private final TransactionTemplate transactionTemplate;
|
private final ThreadPoolTaskExecutor taskExecutor;
|
private final ThreadPoolTaskExecutor runExecutor;
|
|
public EnergyMeterSyncServiceImpl(EnergyMeterDeviceMapper deviceMapper,
|
EnergyMeterMeasurementMapper measurementMapper,
|
EnergyMeterSyncRunMapper runMapper,
|
EnergyMeterSyncTaskMapper taskMapper,
|
EnergyMeterSyncCheckpointMapper checkpointMapper,
|
MeterProviderClient provider,
|
EnergyMeterProperties properties,
|
PlatformTransactionManager transactionManager,
|
@Qualifier("energyMeterSyncExecutor") ThreadPoolTaskExecutor taskExecutor,
|
@Qualifier("energyMeterRunExecutor") ThreadPoolTaskExecutor runExecutor) {
|
this.deviceMapper = deviceMapper;
|
this.measurementMapper = measurementMapper;
|
this.runMapper = runMapper;
|
this.taskMapper = taskMapper;
|
this.checkpointMapper = checkpointMapper;
|
this.provider = provider;
|
this.properties = properties;
|
this.transactionTemplate = new TransactionTemplate(transactionManager);
|
this.taskExecutor = taskExecutor;
|
this.runExecutor = runExecutor;
|
}
|
|
/* ------------------------------------------------------------------ */
|
/* 批次创建 */
|
/* ------------------------------------------------------------------ */
|
|
@Override
|
public Long triggerScheduledSync() {
|
if (!provider.mockMode() && !provider.hasApiKey()) {
|
log.warn("未配置第三方授权码（NH_ENERGY_API_KEY），跳过每日自动同步");
|
return null;
|
}
|
LocalDate today = EnergyMeterSupport.today();
|
LocalDate target = today.minusDays(1);
|
String runKey = EnergyMeterConstants.AUTO_RUN_KEY_PREFIX
|
+ today.format(DateTimeFormatter.BASIC_ISO_DATE);
|
EnergyMeterSyncRun existing = runMapper.selectByRunKey(runKey);
|
if (existing != null) {
|
log.info("当天自动同步批次已存在，跳过重复触发 runKey={}", runKey);
|
return existing.getId();
|
}
|
|
EnergyMeterSyncRun run = newRun(EnergyMeterConstants.RUN_SCHEDULED, runKey, today, target, target);
|
try {
|
runMapper.insert(run);
|
} catch (DuplicateKeyException e) {
|
EnergyMeterSyncRun duplicated = runMapper.selectByRunKey(runKey);
|
log.info("并发触发自动同步，复用已有批次 runKey={}", runKey);
|
return duplicated == null ? null : duplicated.getId();
|
}
|
|
List<EnergyMeterDevice> devices = deviceMapper.selectScannableDevices();
|
LocalDate minStart = target;
|
int taskCount = 0;
|
for (EnergyMeterDevice device : devices) {
|
LocalDate start = resolveAutoStart(device, target);
|
if (start == null) {
|
continue;
|
}
|
if (start.isBefore(minStart)) {
|
minStart = start;
|
}
|
taskCount += createDeviceTasks(run.getId(), device, start, target, properties.isRawReadingEnabled());
|
}
|
applyRunRange(run, minStart, target, taskCount);
|
log.info("自动同步批次创建完成 runKey={} 设备数={} 任务数={}", runKey, devices.size(), taskCount);
|
|
if (taskCount == 0) {
|
EnergyMeterSyncRun finished = new EnergyMeterSyncRun();
|
finished.setId(run.getId());
|
finished.setStatus(EnergyMeterConstants.RUN_SUCCESS);
|
finished.setFinishedAt(EnergyMeterSupport.now());
|
runMapper.updateById(finished);
|
return run.getId();
|
}
|
executeRunAfterCommit(run.getId());
|
return run.getId();
|
}
|
|
@Override
|
public void triggerBackfill(EnergyMeterDevice device, LocalDate start, LocalDate end) {
|
if (device == null || start == null || end == null || start.isAfter(end)) {
|
return;
|
}
|
if (EnergyMeterConstants.isManual(device.getSourceType())) {
|
log.info("手工抄表设备不创建补采批次 meterId={}", device.getId());
|
return;
|
}
|
if (!provider.mockMode() && !provider.hasApiKey()) {
|
log.warn("未配置第三方授权码，跳过补采 meterId={}", device.getId());
|
return;
|
}
|
EnergyMeterSyncRun run = newRun(EnergyMeterConstants.RUN_BACKFILL,
|
"BACKFILL:" + UUID.randomUUID(), null, start, end);
|
runMapper.insert(run);
|
int taskCount = createDeviceTasks(run.getId(), device, start, end, properties.isRawReadingEnabled());
|
applyRunRange(run, start, end, taskCount);
|
log.info("历史补采批次创建完成 meterId={} range={}~{} tasks={}", device.getId(), start, end, taskCount);
|
if (taskCount > 0) {
|
executeRunAfterCommit(run.getId());
|
}
|
}
|
|
@Override
|
public Long triggerBackfillGap(EnergyMeterDevice device) {
|
if (device == null) {
|
return null;
|
}
|
if (EnergyMeterConstants.isManual(device.getSourceType())) {
|
log.info("手工抄表设备不参与缺口补采 meterId={}", device.getId());
|
return null;
|
}
|
if (!provider.mockMode() && !provider.hasApiKey()) {
|
log.warn("未配置第三方授权码，跳过缺口补采 meterId={}", device.getId());
|
return null;
|
}
|
LocalDate target = EnergyMeterSupport.today().minusDays(1);
|
LocalDate base = device.getSyncStartDate() != null
|
? device.getSyncStartDate()
|
: LocalDate.of(target.getYear() - 1, 1, 1);
|
EnergyMeterSyncCheckpoint checkpoint = checkpointMapper.selectUnit(device.getId(),
|
EnergyMeterConstants.usageSource(device.getMeterType()), EnergyMeterConstants.GRANULARITY_DAY);
|
LocalDate start = base;
|
if (checkpoint != null && checkpoint.getLastCompleteDate() != null) {
|
start = checkpoint.getLastCompleteDate().plusDays(1);
|
}
|
if (start.isBefore(base)) {
|
start = base;
|
}
|
if (start.isAfter(target)) {
|
log.info("设备 {} 无缺口需要补采", device.getId());
|
return null;
|
}
|
EnergyMeterSyncRun run = newRun(EnergyMeterConstants.RUN_BACKFILL,
|
"BACKFILL:" + UUID.randomUUID(), null, start, target);
|
runMapper.insert(run);
|
int taskCount = createDeviceTasks(run.getId(), device, start, target, properties.isRawReadingEnabled());
|
applyRunRange(run, start, target, taskCount);
|
if (taskCount > 0) {
|
executeRunAfterCommit(run.getId());
|
}
|
return run.getId();
|
}
|
|
@Override
|
public Long triggerManual(List<Long> meterIds, LocalDate start, LocalDate end, Boolean includeReadings) {
|
if (!provider.mockMode() && !provider.hasApiKey()) {
|
throw new ServiceException("未配置第三方授权码（环境变量 NH_ENERGY_API_KEY），无法查表");
|
}
|
if (meterIds == null || meterIds.isEmpty()) {
|
throw new ServiceException("请选择至少一台设备");
|
}
|
// 手动查表允许查到今天（当天用量为截至此刻的累计值），只挡未来日期
|
LocalDate today = EnergyMeterSupport.today();
|
LocalDate rangeEnd = end != null ? end : today;
|
if (rangeEnd.isAfter(today)) {
|
rangeEnd = today;
|
}
|
LocalDate rangeStart = start != null ? start : rangeEnd.minusDays(2);
|
if (rangeStart.isAfter(rangeEnd)) {
|
throw new ServiceException("开始日期不能晚于结束日期");
|
}
|
if (ChronoUnit.DAYS.between(rangeStart, rangeEnd) > EnergyMeterConstants.MAX_RANGE_DAYS) {
|
throw new ServiceException("查询跨度超出上限，请缩小日期范围");
|
}
|
|
List<EnergyMeterDevice> devices = new ArrayList<>();
|
for (Long meterId : new LinkedHashSet<>(meterIds)) {
|
EnergyMeterDevice device = deviceMapper.selectById(meterId);
|
if (device == null) {
|
throw new ServiceException("设备不存在：" + meterId);
|
}
|
Long tenantId = EnergyMeterSupport.currentTenantId();
|
if (tenantId != null && device.getTenantId() != null && !tenantId.equals(device.getTenantId())) {
|
throw new ServiceException("无权操作设备：" + meterId);
|
}
|
if (!EnergyMeterConstants.BIND_VERIFIED.equals(device.getBindStatus())) {
|
throw new ServiceException("设备未通过第三方档案验证，不能查表：" + meterId);
|
}
|
if (EnergyMeterConstants.isManual(device.getSourceType())) {
|
throw new ServiceException("手工抄表设备无需查表，请在表具台账中录入表码："
|
+ StringUtils.defaultIfBlank(device.getDisplayName(), device.getMeterAddress()));
|
}
|
devices.add(device);
|
}
|
|
EnergyMeterSyncRun run = newRun(EnergyMeterConstants.RUN_MANUAL,
|
"MANUAL:" + UUID.randomUUID(), null, rangeStart, rangeEnd);
|
runMapper.insert(run);
|
boolean withReadings = includeReadings == null ? properties.isRawReadingEnabled() : includeReadings;
|
int taskCount = 0;
|
for (EnergyMeterDevice device : devices) {
|
taskCount += createDeviceTasks(run.getId(), device, rangeStart, rangeEnd, withReadings);
|
}
|
applyRunRange(run, rangeStart, rangeEnd, taskCount);
|
log.info("手动查表批次创建完成 runId={} 设备数={} 任务数={} range={}~{}",
|
run.getId(), devices.size(), taskCount, rangeStart, rangeEnd);
|
if (taskCount > 0) {
|
executeRunAfterCommit(run.getId());
|
}
|
return run.getId();
|
}
|
|
/* ------------------------------------------------------------------ */
|
/* 批次执行 */
|
/* ------------------------------------------------------------------ */
|
|
@Override
|
public void executeRunAsync(Long runId) {
|
if (runId == null) {
|
return;
|
}
|
runExecutor.execute(() -> {
|
try {
|
processRun(runId);
|
} catch (Throwable e) {
|
log.error("同步批次执行异常 runId=" + runId, e);
|
}
|
});
|
}
|
|
/**
|
* 在事务提交后再开始执行，避免异步线程读不到尚未提交的任务记录。
|
*/
|
private void executeRunAfterCommit(Long runId) {
|
if (runId == null) {
|
return;
|
}
|
if (TransactionSynchronizationManager.isSynchronizationActive()) {
|
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
|
@Override
|
public void afterCommit() {
|
executeRunAsync(runId);
|
}
|
});
|
} else {
|
executeRunAsync(runId);
|
}
|
}
|
|
private void processRun(Long runId) {
|
EnergyMeterSyncRun run = runMapper.selectById(runId);
|
if (run == null) {
|
return;
|
}
|
if (EnergyMeterConstants.RUN_PENDING.equals(run.getStatus())) {
|
EnergyMeterSyncRun running = new EnergyMeterSyncRun();
|
running.setId(runId);
|
running.setStatus(EnergyMeterConstants.RUN_RUNNING);
|
running.setStartedAt(EnergyMeterSupport.now());
|
runMapper.updateById(running);
|
}
|
|
LocalDateTime now = EnergyMeterSupport.now();
|
// 同一设备、同一接口、同一粒度的分片按日期顺序串行执行，保证连续进度不会被乱序推进；
|
// 不同设备/接口之间仍然并行，受线程池并发数限制。
|
Map<String, List<EnergyMeterSyncTask>> groups = new LinkedHashMap<>();
|
for (EnergyMeterSyncTask task : taskMapper.selectByRunId(runId)) {
|
if (!isRunnable(task, now)) {
|
continue;
|
}
|
String key = task.getMeterId() + "|" + task.getSourceApi() + "|" + task.getGranularity();
|
groups.computeIfAbsent(key, k -> new ArrayList<>()).add(task);
|
}
|
List<CompletableFuture<Void>> futures = new ArrayList<>();
|
for (List<EnergyMeterSyncTask> group : groups.values()) {
|
group.sort(Comparator.comparing(EnergyMeterSyncTask::getRangeStart,
|
Comparator.nullsFirst(LocalDate::compareTo)));
|
futures.add(CompletableFuture.runAsync(() -> {
|
for (EnergyMeterSyncTask task : group) {
|
executeTask(task.getId());
|
}
|
}, taskExecutor));
|
}
|
if (!futures.isEmpty()) {
|
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
|
}
|
finalizeRun(runId);
|
}
|
|
private boolean isRunnable(EnergyMeterSyncTask task, LocalDateTime now) {
|
if (EnergyMeterConstants.TASK_PENDING.equals(task.getStatus())) {
|
return true;
|
}
|
if (EnergyMeterConstants.TASK_RETRY.equals(task.getStatus())) {
|
return task.getNextRetryAt() == null || !task.getNextRetryAt().isAfter(now);
|
}
|
return false;
|
}
|
|
private void executeTask(Long taskId) {
|
LocalDateTime now = EnergyMeterSupport.now();
|
if (taskMapper.claim(taskId, now) == 0) {
|
return;
|
}
|
EnergyMeterSyncTask task = taskMapper.selectById(taskId);
|
EnergyMeterDevice device = deviceMapper.selectById(task.getMeterId());
|
if (device == null) {
|
markTaskFailed(task, "DEVICE_MISSING", "本地设备不存在，无法采集");
|
return;
|
}
|
try {
|
FetchResult result = fetch(device, task);
|
persist(device, task, result);
|
} catch (MeterProviderException e) {
|
markTaskFailed(task, e.getCode(), e.getMessage());
|
} catch (Exception e) {
|
log.error("同步任务异常 taskId=" + taskId, e);
|
markTaskFailed(task, EnergyMeterConstants.ERROR_UPSTREAM, e.getMessage());
|
}
|
}
|
|
private FetchResult fetch(EnergyMeterDevice device, EnergyMeterSyncTask task) {
|
String meterType = device.getMeterType();
|
String address = device.getMeterAddress();
|
String providerMeterId = device.getProviderMeterId();
|
LocalDate start = task.getRangeStart();
|
LocalDate end = task.getRangeEnd();
|
LocalDateTime collectedAt = EnergyMeterSupport.now();
|
List<EnergyMeterMeasurement> measurements = new ArrayList<>();
|
int fetched = 0;
|
|
if (EnergyMeterConstants.GRANULARITY_DAY.equals(task.getGranularity())) {
|
List<MeterUsagePoint> points = provider.queryDailyUsage(meterType, address, providerMeterId, start, end);
|
fetched = points.size();
|
String metric = EnergyMeterConstants.usageMetric(meterType);
|
String unit = EnergyMeterConstants.unit(meterType);
|
for (MeterUsagePoint point : points) {
|
validateAddress(address, point.getAddress());
|
if (point.getTime() == null || point.getValue() == null) {
|
throw new MeterProviderException(EnergyMeterConstants.ERROR_PARSE, "用量数据缺少时间或数值");
|
}
|
LocalDate day = point.getTime().toLocalDate();
|
if (day.isBefore(start) || day.isAfter(end)) {
|
continue;
|
}
|
EnergyMeterMeasurement measurement = new EnergyMeterMeasurement();
|
measurement.setTenantId(device.getTenantId());
|
measurement.setMeterId(device.getId());
|
measurement.setMetricCode(metric);
|
measurement.setGranularity(EnergyMeterConstants.GRANULARITY_DAY);
|
measurement.setMeasureTime(day.atStartOfDay());
|
measurement.setMetricValue(point.getValue());
|
measurement.setUnit(unit);
|
measurement.setSourceApi(task.getSourceApi());
|
measurement.setSourceRecordTime(point.getTime());
|
measurement.setCollectedAt(collectedAt);
|
measurements.add(measurement);
|
}
|
} else {
|
List<MeterReadingPoint> points = provider.queryReadings(meterType, address, providerMeterId, start, end);
|
fetched = points.size();
|
String metric = EnergyMeterConstants.readingMetric(meterType);
|
String unit = EnergyMeterConstants.unit(meterType);
|
for (MeterReadingPoint point : points) {
|
validateAddress(address, point.getAddress());
|
if (point.getTime() == null || point.getReading() == null) {
|
throw new MeterProviderException(EnergyMeterConstants.ERROR_PARSE, "示值数据缺少时间或数值");
|
}
|
LocalDate day = point.getTime().toLocalDate();
|
if (day.isBefore(start) || day.isAfter(end)) {
|
continue;
|
}
|
EnergyMeterMeasurement measurement = new EnergyMeterMeasurement();
|
measurement.setTenantId(device.getTenantId());
|
measurement.setMeterId(device.getId());
|
measurement.setMetricCode(metric);
|
measurement.setGranularity(EnergyMeterConstants.GRANULARITY_POINT);
|
measurement.setMeasureTime(point.getTime());
|
measurement.setMetricValue(point.getReading());
|
measurement.setUnit(unit);
|
measurement.setSourceApi(task.getSourceApi());
|
measurement.setSourceRecordTime(point.getTime());
|
measurement.setCollectedAt(collectedAt);
|
measurements.add(measurement);
|
}
|
}
|
FetchResult result = new FetchResult();
|
result.fetched = fetched;
|
result.measurements = measurements;
|
result.metricCode = EnergyMeterConstants.GRANULARITY_DAY.equals(task.getGranularity())
|
? EnergyMeterConstants.usageMetric(meterType)
|
: EnergyMeterConstants.readingMetric(meterType);
|
return result;
|
}
|
|
private void validateAddress(String expected, String actual) {
|
if (StringUtils.isNotBlank(actual) && !expected.equals(actual.trim())) {
|
throw new MeterProviderException(EnergyMeterConstants.ERROR_ADDRESS_MISMATCH,
|
"第三方返回的通信地址与本地绑定不一致");
|
}
|
}
|
|
private void persist(EnergyMeterDevice device, EnergyMeterSyncTask task, FetchResult result) {
|
LocalDateTime now = EnergyMeterSupport.now();
|
LocalDateTime rangeStart = task.getRangeStart().atStartOfDay();
|
LocalDateTime rangeEnd = task.getRangeEnd().plusDays(1).atStartOfDay();
|
|
int[] counters = transactionTemplate.execute(txStatus -> {
|
int inserted = 0;
|
int updated = 0;
|
List<EnergyMeterMeasurement> writes = new ArrayList<>();
|
if (!result.measurements.isEmpty()) {
|
Map<LocalDateTime, BigDecimal> existing = new HashMap<>();
|
for (EnergyMeterMeasurement row : measurementMapper.selectExistingPoints(device.getId(),
|
result.metricCode, task.getGranularity(), rangeStart, rangeEnd)) {
|
existing.put(row.getMeasureTime(), row.getMetricValue());
|
}
|
for (EnergyMeterMeasurement measurement : result.measurements) {
|
BigDecimal oldValue = existing.get(measurement.getMeasureTime());
|
if (oldValue == null) {
|
inserted++;
|
writes.add(measurement);
|
} else if (oldValue.compareTo(measurement.getMetricValue()) != 0) {
|
updated++;
|
writes.add(measurement);
|
}
|
}
|
}
|
for (int from = 0; from < writes.size(); from += UPSERT_CHUNK) {
|
int to = Math.min(from + UPSERT_CHUNK, writes.size());
|
measurementMapper.upsertBatch(writes.subList(from, to));
|
}
|
|
String status = result.measurements.isEmpty()
|
? EnergyMeterConstants.TASK_EMPTY : EnergyMeterConstants.TASK_SUCCESS;
|
taskMapper.markFinished(task.getId(), status, result.fetched, inserted, updated, now);
|
deviceMapper.markCollectSuccess(device.getId(), now);
|
advanceCheckpoint(device, task, now);
|
return new int[]{inserted, updated};
|
});
|
int inserted = counters == null ? 0 : counters[0];
|
int updated = counters == null ? 0 : counters[1];
|
log.info("同步任务完成 taskId={} meterId={} 拉取={} 新增={} 修正={}",
|
task.getId(), device.getId(), result.fetched, inserted, updated);
|
}
|
|
private void advanceCheckpoint(EnergyMeterDevice device, EnergyMeterSyncTask task, LocalDateTime now) {
|
LocalDate start = task.getRangeStart();
|
LocalDate end = task.getRangeEnd();
|
// 当天数据仍在滚动累计（手动查表或当天补采会不断覆盖），不能把今天记为“已完成”，
|
// 否则补采缺口判断会误以为今天已经完整。完成进度只推进到昨天及更早。
|
if (!end.isBefore(EnergyMeterSupport.today())) {
|
log.debug("跳过当天进度推进 meterId={} source={} granularity={} end={}",
|
device.getId(), task.getSourceApi(), task.getGranularity(), end);
|
return;
|
}
|
EnergyMeterSyncCheckpoint checkpoint = checkpointMapper.selectUnit(device.getId(),
|
task.getSourceApi(), task.getGranularity());
|
LocalDate existingDate = checkpoint == null ? null : checkpoint.getLastCompleteDate();
|
boolean contiguous = existingDate == null || !start.isAfter(existingDate.plusDays(1));
|
LocalDate lastComplete;
|
if (contiguous) {
|
lastComplete = existingDate == null || end.isAfter(existingDate) ? end : existingDate;
|
} else {
|
lastComplete = existingDate != null ? existingDate : start.minusDays(1);
|
}
|
EnergyMeterSyncCheckpoint upsert = new EnergyMeterSyncCheckpoint();
|
upsert.setTenantId(device.getTenantId());
|
upsert.setMeterId(device.getId());
|
upsert.setSourceApi(task.getSourceApi());
|
upsert.setGranularity(task.getGranularity());
|
upsert.setLastCompleteDate(lastComplete);
|
upsert.setLastSuccessAt(now);
|
upsert.setLastRunId(task.getRunId());
|
checkpointMapper.upsert(upsert);
|
}
|
|
private void markTaskFailed(EnergyMeterSyncTask task, String code, String message) {
|
String safeCode = StringUtils.defaultIfBlank(code, EnergyMeterConstants.ERROR_UPSTREAM);
|
String safeMessage = EnergyMeterSupport.truncate(
|
StringUtils.defaultIfBlank(message, "未知错误"), 450);
|
int attempts = task.getAttemptCount() == null ? 1 : task.getAttemptCount();
|
// max-attempts 表示“最多重试次数”，首次执行之外还可重试 N 次
|
int maxTotalAttempts = Math.max(1, properties.getMaxAttempts()) + 1;
|
boolean retry = attempts < maxTotalAttempts;
|
LocalDateTime now = EnergyMeterSupport.now();
|
LocalDateTime nextRetryAt = null;
|
if (retry) {
|
int index = Math.min(Math.max(attempts - 1, 0), EnergyMeterConstants.RETRY_DELAY_MINUTES.length - 1);
|
nextRetryAt = now.plusMinutes(EnergyMeterConstants.RETRY_DELAY_MINUTES[index]);
|
}
|
String status = retry ? EnergyMeterConstants.TASK_RETRY : EnergyMeterConstants.TASK_FAILED;
|
LocalDateTime finalNextRetryAt = nextRetryAt;
|
transactionTemplate.executeWithoutResult(state -> {
|
taskMapper.markFailed(task.getId(), status, safeCode, safeMessage, finalNextRetryAt, now);
|
deviceMapper.markCollectFailure(task.getMeterId(), safeCode, safeMessage);
|
});
|
log.warn("同步任务失败 taskId={} meterId={} code={} attempts={} 将重试={}",
|
task.getId(), task.getMeterId(), safeCode, attempts, retry);
|
}
|
|
private void finalizeRun(Long runId) {
|
TaskStatusCount count = taskMapper.countByRunId(runId);
|
runMapper.refreshTotals(runId);
|
int total = count == null || count.getTotal() == null ? 0 : count.getTotal();
|
int success = count == null || count.getSuccess() == null ? 0 : count.getSuccess();
|
int empty = count == null || count.getEmpty() == null ? 0 : count.getEmpty();
|
int failed = count == null || count.getFailed() == null ? 0 : count.getFailed();
|
int retry = count == null || count.getRetry() == null ? 0 : count.getRetry();
|
int pending = count == null || count.getPending() == null ? 0 : count.getPending();
|
int running = count == null || count.getRunning() == null ? 0 : count.getRunning();
|
|
boolean finished = pending == 0 && running == 0 && retry == 0;
|
String status;
|
if (!finished) {
|
status = EnergyMeterConstants.RUN_PARTIAL;
|
} else if (failed == 0) {
|
status = EnergyMeterConstants.RUN_SUCCESS;
|
} else if (success + empty > 0) {
|
status = EnergyMeterConstants.RUN_PARTIAL;
|
} else {
|
status = EnergyMeterConstants.RUN_FAILED;
|
}
|
EnergyMeterSyncRun update = new EnergyMeterSyncRun();
|
update.setId(runId);
|
update.setStatus(status);
|
if (finished) {
|
update.setFinishedAt(EnergyMeterSupport.now());
|
}
|
if (failed > 0) {
|
update.setErrorSummary(EnergyMeterSupport.truncate(
|
failed + "/" + total + " 个任务失败，可在批次详情中重试", 500));
|
}
|
runMapper.updateById(update);
|
log.info("同步批次汇总 runId={} status={} total={} success={} empty={} failed={} retry={}",
|
runId, status, total, success, empty, failed, retry);
|
}
|
|
/* ------------------------------------------------------------------ */
|
/* 重试与恢复 */
|
/* ------------------------------------------------------------------ */
|
|
@Override
|
public int recoverDueTasks() {
|
LocalDateTime now = EnergyMeterSupport.now();
|
int timeoutMinutes = Math.max(1, properties.getRunningTimeoutMinutes());
|
LocalDateTime deadline = now.minusMinutes(timeoutMinutes);
|
int handled = 0;
|
int maxTotalAttempts = Math.max(1, properties.getMaxAttempts()) + 1;
|
for (EnergyMeterSyncTask task : taskMapper.selectStaleRunning(deadline, 200)) {
|
int attempts = task.getAttemptCount() == null ? 0 : task.getAttemptCount();
|
if (attempts >= maxTotalAttempts) {
|
transactionTemplate.executeWithoutResult(state -> taskMapper.markFailed(task.getId(),
|
EnergyMeterConstants.TASK_FAILED, EnergyMeterConstants.ERROR_TIMEOUT,
|
"任务执行超时（超过 " + timeoutMinutes + " 分钟）", null, now));
|
} else {
|
taskMapper.resetToRetry(task.getId(), now, now, "任务执行超时，自动转入重试");
|
}
|
handled++;
|
}
|
|
List<EnergyMeterSyncTask> dueTasks = taskMapper.selectDueRetry(now, 200);
|
Set<Long> runIds = dueTasks.stream().map(EnergyMeterSyncTask::getRunId)
|
.collect(Collectors.toCollection(LinkedHashSet::new));
|
for (Long runId : runIds) {
|
try {
|
processRun(runId);
|
} catch (Exception e) {
|
log.error("恢复任务处理批次失败 runId=" + runId, e);
|
}
|
}
|
handled += dueTasks.size();
|
|
// 崩溃兜底：批次停在 RUNNING 但任务都已结束（例如汇总阶段异常退出）时重新汇总
|
for (EnergyMeterSyncRun stuck : runMapper.selectStuckRuns(now.minusMinutes(5), 50)) {
|
try {
|
finalizeRun(stuck.getId());
|
handled++;
|
log.info("兜底汇总滞留批次 runId={}", stuck.getId());
|
} catch (Exception e) {
|
log.error("兜底汇总批次失败 runId=" + stuck.getId(), e);
|
}
|
}
|
if (handled > 0) {
|
log.info("同步恢复作业处理 {} 个任务/批次", handled);
|
}
|
return handled;
|
}
|
|
/* ------------------------------------------------------------------ */
|
/* 内部工具 */
|
/* ------------------------------------------------------------------ */
|
|
private EnergyMeterSyncRun newRun(String triggerType, String runKey, LocalDate scheduleDate,
|
LocalDate start, LocalDate end) {
|
EnergyMeterSyncRun run = new EnergyMeterSyncRun();
|
run.setRunKey(runKey);
|
run.setTriggerType(triggerType);
|
run.setScheduleDate(scheduleDate);
|
run.setRangeStart(start);
|
run.setRangeEnd(end);
|
run.setStatus(EnergyMeterConstants.RUN_PENDING);
|
run.setTotalTasks(0);
|
run.setSuccessTasks(0);
|
run.setEmptyTasks(0);
|
run.setFailedTasks(0);
|
run.setCreateUser(EnergyMeterSupport.currentUserId());
|
return run;
|
}
|
|
private void applyRunRange(EnergyMeterSyncRun run, LocalDate start, LocalDate end, int taskCount) {
|
EnergyMeterSyncRun update = new EnergyMeterSyncRun();
|
update.setId(run.getId());
|
update.setRangeStart(start);
|
update.setRangeEnd(end);
|
update.setTotalTasks(taskCount);
|
runMapper.updateById(update);
|
}
|
|
private int createDeviceTasks(Long runId, EnergyMeterDevice device, LocalDate start, LocalDate end,
|
boolean includeReadings) {
|
if (EnergyMeterConstants.isManual(device.getSourceType())) {
|
return 0;
|
}
|
int created = createTasksFor(runId, device,
|
EnergyMeterConstants.usageSource(device.getMeterType()),
|
EnergyMeterConstants.GRANULARITY_DAY, start, end,
|
Math.max(1, properties.getMaxShardDays()));
|
if (includeReadings && !EnergyMeterConstants.isGas(device.getMeterType())) {
|
// 示值接口的查询跨度限制比用量接口更严（集社云为 15 天），必须用更小的分片。
|
// 气表示值接口只支持 mi15（15 分钟）粒度，96 点/天，数据量过大，暂不创建示值任务。
|
created += createTasksFor(runId, device,
|
EnergyMeterConstants.readingSource(device.getMeterType()),
|
EnergyMeterConstants.GRANULARITY_POINT, start, end,
|
Math.max(1, properties.getMaxReadingShardDays()));
|
}
|
return created;
|
}
|
|
private int createTasksFor(Long runId, EnergyMeterDevice device, String sourceApi, String granularity,
|
LocalDate start, LocalDate end, int shardDays) {
|
int created = 0;
|
LocalDate cursor = start;
|
while (!cursor.isAfter(end)) {
|
LocalDate shardEnd = cursor.plusDays(shardDays - 1L);
|
if (shardEnd.isAfter(end)) {
|
shardEnd = end;
|
}
|
created += taskMapper.insertIgnore(buildTask(runId, device, sourceApi, granularity, cursor, shardEnd));
|
cursor = shardEnd.plusDays(1);
|
}
|
return created;
|
}
|
|
private EnergyMeterSyncTask buildTask(Long runId, EnergyMeterDevice device, String sourceApi,
|
String granularity, LocalDate start, LocalDate end) {
|
EnergyMeterSyncTask task = new EnergyMeterSyncTask();
|
task.setRunId(runId);
|
task.setTenantId(device.getTenantId());
|
task.setMeterId(device.getId());
|
task.setSourceApi(sourceApi);
|
task.setGranularity(granularity);
|
task.setRangeStart(start);
|
task.setRangeEnd(end);
|
task.setStatus(EnergyMeterConstants.TASK_PENDING);
|
return task;
|
}
|
|
/**
|
* 自动同步起点：优先补齐进度缺口，同时始终回查最近 N 天（默认 3 天）。
|
*/
|
private LocalDate resolveAutoStart(EnergyMeterDevice device, LocalDate target) {
|
LocalDate base = device.getSyncStartDate() != null
|
? device.getSyncStartDate()
|
: LocalDate.of(target.getYear() - 1, 1, 1);
|
LocalDate start = base;
|
EnergyMeterSyncCheckpoint checkpoint = checkpointMapper.selectUnit(device.getId(),
|
EnergyMeterConstants.usageSource(device.getMeterType()), EnergyMeterConstants.GRANULARITY_DAY);
|
if (checkpoint != null && checkpoint.getLastCompleteDate() != null) {
|
LocalDate next = checkpoint.getLastCompleteDate().plusDays(1);
|
if (next.isAfter(start)) {
|
start = next;
|
}
|
}
|
int lookback = Math.max(1, properties.getLookbackDays());
|
LocalDate lookbackStart = target.minusDays(lookback - 1L);
|
if (start.isAfter(lookbackStart)) {
|
start = lookbackStart;
|
}
|
if (start.isBefore(base)) {
|
start = base;
|
}
|
if (start.isAfter(target)) {
|
return null;
|
}
|
return start;
|
}
|
|
/** 一次任务的拉取结果。 */
|
private static class FetchResult {
|
private int fetched;
|
private String metricCode;
|
private List<EnergyMeterMeasurement> measurements = new ArrayList<>();
|
}
|
}
|
