13 changed files with 648 additions and 128 deletions
@ -0,0 +1,16 @@
|
||||
package com.mh.user.service; |
||||
|
||||
import java.text.ParseException; |
||||
|
||||
/** |
||||
* @author LJF |
||||
* @version 1.0 |
||||
* @project mh_esi |
||||
* @description pro_data_result修改java |
||||
* @date 2025-08-26 09:08:15 |
||||
*/ |
||||
public interface ProDataResultService { |
||||
|
||||
void processDailySummary(String curDate, String projectId) throws ParseException; |
||||
|
||||
} |
||||
@ -0,0 +1,328 @@
|
||||
package com.mh.user.service.impl; |
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; |
||||
import com.mh.user.entity.DataResultChEntity; |
||||
import com.mh.user.entity.DataResultFifteenMiEntity; |
||||
import com.mh.user.entity.DataResultFiveMiEntity; |
||||
import com.mh.user.entity.DataResultOneMiEntity; |
||||
import com.mh.user.mapper.DataResultChMapper; |
||||
import com.mh.user.mapper.DataResultFifteenMiMapper; |
||||
import com.mh.user.mapper.DataResultFiveMiMapper; |
||||
import com.mh.user.mapper.DataResultOneMiMapper; |
||||
import com.mh.user.service.ProDataResultService; |
||||
import lombok.extern.slf4j.Slf4j; |
||||
import org.springframework.beans.BeanUtils; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.stereotype.Service; |
||||
import org.springframework.transaction.annotation.Transactional; |
||||
|
||||
import java.sql.Timestamp; |
||||
import java.text.ParseException; |
||||
import java.text.SimpleDateFormat; |
||||
import java.util.*; |
||||
import java.util.concurrent.ConcurrentHashMap; |
||||
import java.util.stream.Collectors; |
||||
|
||||
/** |
||||
* @author LJF |
||||
* @version 1.0 |
||||
* @project mh_esi |
||||
* @description pro_data_result改成java |
||||
* @date 2025-08-26 09:07:39 |
||||
*/ |
||||
@Slf4j |
||||
@Service |
||||
@Transactional(rollbackFor = Exception.class) |
||||
public class ProDataResultServiceImpl implements ProDataResultService { |
||||
|
||||
private static final List<String> DEVICE_CODES = Arrays.asList("0020", "0073", "0075", "0006", "0010", "0012", "0014"); |
||||
|
||||
@Autowired |
||||
private DataResultChMapper dataResultChMapper; |
||||
@Autowired |
||||
private DataResultOneMiMapper oneMiMapper; |
||||
@Autowired |
||||
private DataResultFiveMiMapper fiveMiMapper; |
||||
@Autowired |
||||
private DataResultFifteenMiMapper fifteenMiMapper; |
||||
|
||||
@Override |
||||
public void processDailySummary(String curDateStr, String projectId) throws ParseException { |
||||
// 时间格式标准化
|
||||
SimpleDateFormat inputSdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); |
||||
Date curDate = inputSdf.parse(curDateStr); |
||||
SimpleDateFormat outputSdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:00"); |
||||
String standardDate = outputSdf.format(curDate); |
||||
log.info("处理日期：{}", standardDate); |
||||
|
||||
// 计算时间窗口参数
|
||||
int minute = curDate.getMinutes(); |
||||
int intervalMinute = (minute / 5) * 5; |
||||
String minuteStr = String.format("%02d", intervalMinute); |
||||
log.info("分钟区间：{}", minuteStr); |
||||
|
||||
// 遍历设备编码
|
||||
for (String deviceCode : DEVICE_CODES) { |
||||
log.info("处理设备编码：{}", deviceCode); |
||||
|
||||
// 检查是否存在待处理数据
|
||||
Map<String, Object> countParams = new HashMap<>(); |
||||
// countParams.put("start", standardDate);
|
||||
// countParams.put("end", addMinutes(standardDate, 5));
|
||||
countParams.put("register_addr", deviceCode); |
||||
countParams.put("project_id", projectId); |
||||
int recordCount = Math.toIntExact(dataResultChMapper.selectCount(new QueryWrapper<DataResultChEntity>() |
||||
.allEq(countParams) |
||||
.ge("cur_date", standardDate) |
||||
.lt("cur_date", addMinutes(standardDate, 5)))); |
||||
|
||||
if (recordCount == 0) continue; |
||||
|
||||
// 获取当前设备编码下的所有设备地址
|
||||
List<String> deviceAddrs = dataResultChMapper.selectObjs(new QueryWrapper<DataResultChEntity>() |
||||
.select("DISTINCT device_addr") |
||||
.eq("register_addr", deviceCode) |
||||
.eq("project_id", projectId) |
||||
.between("cur_date", standardDate, addMinutes(standardDate, 5))) |
||||
.stream().map(Object::toString).collect(Collectors.toList()); |
||||
|
||||
// 处理每个设备地址
|
||||
for (String deviceAddr : deviceAddrs) { |
||||
// processIntervalData(standardDate, deviceCode, projectId, deviceAddr, 1);
|
||||
processIntervalData(standardDate, deviceCode, projectId, deviceAddr, 5, minuteStr); |
||||
processIntervalData(standardDate, deviceCode, projectId, deviceAddr, 15, minuteStr); |
||||
} |
||||
} |
||||
} |
||||
|
||||
private void processIntervalData(String baseDate, String registerAddr, String projectId, |
||||
String deviceAddr, int interval, String... extraParams) throws ParseException { |
||||
// 计算时间窗口
|
||||
String startTime = baseDate; |
||||
String endTime = addMinutes(baseDate, interval); |
||||
|
||||
// 获取最新数据
|
||||
Map<String, Object> params = new HashMap<>(); |
||||
params.put("start", startTime); |
||||
params.put("end", endTime); |
||||
params.put("registerAddr", registerAddr); |
||||
params.put("projectId", projectId); |
||||
params.put("deviceAddr", deviceAddr); |
||||
DataResultChEntity latest = dataResultChMapper.selectLatestByCondition(params); |
||||
|
||||
if (latest == null) return; |
||||
|
||||
// 数值有效性校验
|
||||
String curValue = isNumeric(latest.getCurValue()) ? latest.getCurValue() : "0.00"; |
||||
if (!isNumeric(latest.getCurValue())) { |
||||
log.info("无效数值，已替换为0.00：{}", latest.getCurValue()); |
||||
} |
||||
|
||||
// 计算汇总时间点
|
||||
String summaryTime = extraParams.length > 0 ? |
||||
baseDate.substring(0, 13) + ":" + extraParams[0] + ":00" : |
||||
new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(latest.getCurDate()); |
||||
|
||||
// 处理不同时间间隔的汇总
|
||||
switch (interval) { |
||||
case 1: |
||||
handleOneMinute(latest, deviceAddr, projectId, summaryTime, curValue); |
||||
break; |
||||
case 5: |
||||
handleFiveMinute(latest, deviceAddr, projectId, summaryTime, curValue); |
||||
break; |
||||
case 15: |
||||
handleFifteenMinute(latest, deviceAddr, projectId, summaryTime, curValue); |
||||
break; |
||||
} |
||||
} |
||||
|
||||
// 在类中添加静态锁对象
|
||||
private static final Map<String, Object> LOCK_MAP = new ConcurrentHashMap<>(); |
||||
|
||||
private void handleOneMinute(DataResultChEntity source, String deviceAddr, String projectId, |
||||
String summaryTime, String curValue) { |
||||
// 创建针对特定设备和项目的锁key
|
||||
String lockKey = deviceAddr + "_" + projectId; |
||||
Object lockObject = LOCK_MAP.computeIfAbsent(lockKey, k -> new Object()); |
||||
|
||||
try { |
||||
// 使用特定锁，减少锁竞争
|
||||
synchronized (lockObject) { |
||||
// 添加重试机制避免死锁
|
||||
int retryCount = 0; |
||||
final int maxRetries = 3; |
||||
|
||||
while (retryCount < maxRetries) { |
||||
try { |
||||
// 查询历史数据
|
||||
int count = oneMiMapper.selectCountByTimeDeviceProject(summaryTime, deviceAddr, projectId, source.getRegisterAddr(), source.getDeviceType()); |
||||
if (count == 0) { |
||||
// 插入新数据
|
||||
DataResultOneMiEntity record = new DataResultOneMiEntity(); |
||||
BeanUtils.copyProperties(source, record); |
||||
record.setCurDate(Timestamp.valueOf(summaryTime)); |
||||
record.setCurValue(curValue); |
||||
oneMiMapper.saveDataResultOneMi(record); |
||||
} |
||||
break; // 成功执行则跳出循环
|
||||
|
||||
} catch (Exception e) { |
||||
retryCount++; |
||||
if (retryCount >= maxRetries) { |
||||
throw e; // 达到最大重试次数后抛出异常
|
||||
} |
||||
|
||||
// 检查是否为死锁异常
|
||||
if (e.getMessage() != null && |
||||
(e.getMessage().contains("死锁") || e.getMessage().contains("deadlock"))) { |
||||
log.warn("检测到死锁，进行第{}次重试，设备地址:{}，项目ID:{}", |
||||
retryCount, deviceAddr, projectId); |
||||
// 随机延迟后重试
|
||||
try { |
||||
Thread.sleep(new Random().nextInt(200) + 100); |
||||
} catch (InterruptedException ie) { |
||||
Thread.currentThread().interrupt(); |
||||
throw new RuntimeException(ie); |
||||
} |
||||
} else { |
||||
throw e; // 非死锁异常直接抛出
|
||||
} |
||||
} |
||||
} |
||||
} |
||||
} finally { |
||||
// 清理长时间不用的锁对象，避免内存泄漏
|
||||
cleanupLocks(); |
||||
} |
||||
} |
||||
|
||||
// 清理锁对象的方法
|
||||
private void cleanupLocks() { |
||||
// 简单的清理策略：当锁数量超过一定阈值时，清空所有锁
|
||||
// 实际项目中可以根据需要实现更复杂的清理逻辑
|
||||
if (LOCK_MAP.size() > 1000) { |
||||
synchronized (LOCK_MAP) { |
||||
if (LOCK_MAP.size() > 1000) { |
||||
LOCK_MAP.clear(); |
||||
} |
||||
} |
||||
} |
||||
} |
||||
|
||||
private void handleFiveMinute(DataResultChEntity source, String deviceAddr, String projectId, |
||||
String summaryTime, String curValue) { |
||||
// 创建针对特定设备和项目的锁key
|
||||
String lockKey = "five_" + deviceAddr + "_" + projectId; |
||||
Object lockObject = LOCK_MAP.computeIfAbsent(lockKey, k -> new Object()); |
||||
|
||||
try { |
||||
// 使用特定锁，减少锁竞争
|
||||
synchronized (lockObject) { |
||||
// 添加重试机制避免死锁
|
||||
executeWithRetry(() -> { |
||||
Long existId = fiveMiMapper.selectExistId(summaryTime, deviceAddr, projectId); |
||||
|
||||
DataResultFiveMiEntity record = new DataResultFiveMiEntity(); |
||||
BeanUtils.copyProperties(source, record); |
||||
record.setCurDate(Timestamp.valueOf(summaryTime)); |
||||
record.setCurValue(curValue); |
||||
|
||||
if (existId != null) { |
||||
fiveMiMapper.updateByIdCustomize(existId, curValue, source.getGrade()); |
||||
} else { |
||||
fiveMiMapper.saveDataResultFiveMi(record); |
||||
} |
||||
}); |
||||
} |
||||
} finally { |
||||
// 清理长时间不用的锁对象，避免内存泄漏
|
||||
cleanupLocks(); |
||||
} |
||||
} |
||||
|
||||
private void handleFifteenMinute(DataResultChEntity source, String deviceAddr, String projectId, |
||||
String summaryTime, String curValue) { |
||||
// 创建针对特定设备和项目的锁key
|
||||
String lockKey = "fifteen_" + deviceAddr + "_" + projectId; |
||||
Object lockObject = LOCK_MAP.computeIfAbsent(lockKey, k -> new Object()); |
||||
|
||||
try { |
||||
// 使用特定锁，减少锁竞争
|
||||
synchronized (lockObject) { |
||||
// 添加重试机制避免死锁
|
||||
executeWithRetry(() -> { |
||||
Long existId = fifteenMiMapper.selectExistId(summaryTime, deviceAddr, projectId); |
||||
|
||||
DataResultFifteenMiEntity record = new DataResultFifteenMiEntity(); |
||||
BeanUtils.copyProperties(source, record); |
||||
record.setCurDate(Timestamp.valueOf(summaryTime)); |
||||
record.setCurValue(curValue); |
||||
|
||||
if (existId != null) { |
||||
fifteenMiMapper.updateByIdCustomize(existId, curValue, source.getGrade()); |
||||
} else { |
||||
fifteenMiMapper.saveDataResultFifteenMi(record); |
||||
} |
||||
}); |
||||
} |
||||
} finally { |
||||
// 清理长时间不用的锁对象，避免内存泄漏
|
||||
cleanupLocks(); |
||||
} |
||||
} |
||||
|
||||
// 通用重试方法
|
||||
private void executeWithRetry(Runnable operation) { |
||||
int retryCount = 0; |
||||
final int maxRetries = 3; |
||||
|
||||
while (retryCount < maxRetries) { |
||||
try { |
||||
operation.run(); |
||||
break; |
||||
} catch (Exception e) { |
||||
retryCount++; |
||||
if (retryCount >= maxRetries) { |
||||
throw new RuntimeException("操作失败，已重试" + maxRetries + "次", e); |
||||
} |
||||
|
||||
// 检查是否为死锁异常
|
||||
if (isDeadlockException(e)) { |
||||
log.warn("检测到死锁，进行第{}次重试", retryCount); |
||||
// 随机延迟后重试
|
||||
try { |
||||
Thread.sleep(new Random().nextInt(200) + 100); |
||||
} catch (InterruptedException ie) { |
||||
Thread.currentThread().interrupt(); |
||||
throw new RuntimeException(ie); |
||||
} |
||||
} else { |
||||
throw new RuntimeException("非死锁异常", e); // 非死锁异常直接抛出
|
||||
} |
||||
} |
||||
} |
||||
} |
||||
|
||||
// 判断是否为死锁异常
|
||||
private boolean isDeadlockException(Exception e) { |
||||
if (e.getMessage() != null) { |
||||
String message = e.getMessage().toLowerCase(); |
||||
return message.contains("死锁") || message.contains("deadlock"); |
||||
} |
||||
return false; |
||||
} |
||||
|
||||
|
||||
private String addMinutes(String baseDate, int minutes) throws ParseException { |
||||
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); |
||||
Date date = sdf.parse(baseDate); |
||||
date.setTime(date.getTime() + minutes * 60 * 1000L); |
||||
return sdf.format(date); |
||||
} |
||||
|
||||
private boolean isNumeric(String str) { |
||||
return str != null && str.matches("-?\\d+(\\.\\d+)?"); |
||||
} |
||||
|
||||
} |
||||
Loading…
Reference in new issue