|
|
|
|
@ -19,23 +19,27 @@ import com.mh.user.service.GatewayManageService;
|
|
|
|
|
import com.mh.user.service.mqtt.service.IEventsService; |
|
|
|
|
import com.mh.user.strategy.DeviceStrategy; |
|
|
|
|
import com.mh.user.strategy.DeviceStrategyFactory; |
|
|
|
|
import com.mh.user.utils.CacheUtil; |
|
|
|
|
import com.mh.user.utils.DateUtil; |
|
|
|
|
import com.mh.user.utils.SpringBeanUtil; |
|
|
|
|
import io.netty.util.CharsetUtil; |
|
|
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
|
|
import org.springframework.beans.factory.annotation.Qualifier; |
|
|
|
|
import org.springframework.context.ApplicationContext; |
|
|
|
|
import org.springframework.integration.annotation.ServiceActivator; |
|
|
|
|
import org.springframework.messaging.MessageHeaders; |
|
|
|
|
import org.springframework.stereotype.Service; |
|
|
|
|
|
|
|
|
|
import javax.annotation.PostConstruct; |
|
|
|
|
import javax.annotation.PreDestroy; |
|
|
|
|
import java.io.IOException; |
|
|
|
|
import java.math.BigDecimal; |
|
|
|
|
import java.util.ArrayList; |
|
|
|
|
import java.util.HashMap; |
|
|
|
|
import java.util.List; |
|
|
|
|
import java.util.Map; |
|
|
|
|
import java.util.Objects; |
|
|
|
|
import java.util.Optional; |
|
|
|
|
import java.util.concurrent.*; |
|
|
|
|
import java.util.concurrent.atomic.AtomicInteger; |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* @author LJF |
|
|
|
|
@ -62,8 +66,129 @@ public class EventsServiceImpl implements IEventsService {
|
|
|
|
|
private DeviceInstallService deviceInstallService; |
|
|
|
|
|
|
|
|
|
@Autowired |
|
|
|
|
@Qualifier("caffeineCache") |
|
|
|
|
private Cache caffeineCache; |
|
|
|
|
|
|
|
|
|
// 常量定义
|
|
|
|
|
private static final int BATCH_SIZE = 100; |
|
|
|
|
private static final long TIME_INTERVAL_THRESHOLD_MS = 150000; // 150秒
|
|
|
|
|
|
|
|
|
|
// 线程池配置
|
|
|
|
|
private static final int CORE_POOL_SIZE = 3; |
|
|
|
|
private static final int MAX_POOL_SIZE = 5; |
|
|
|
|
private static final int QUEUE_CAPACITY = 100; |
|
|
|
|
private static final long KEEP_ALIVE_TIME = 30L; |
|
|
|
|
|
|
|
|
|
// 缓存配置
|
|
|
|
|
private static final int TIME_CACHE_MAX_SIZE = 1000; |
|
|
|
|
private static final int TIME_CACHE_EXPIRE_MINUTES = 30; |
|
|
|
|
private static final int DEVICE_CACHE_MAX_SIZE = 500; |
|
|
|
|
private static final int DEVICE_CACHE_EXPIRE_MINUTES = 60; |
|
|
|
|
|
|
|
|
|
// 线程池
|
|
|
|
|
private ExecutorService deviceAnalysisExecutor; |
|
|
|
|
|
|
|
|
|
// 本地缓存，使用带过期时间的缓存
|
|
|
|
|
private final ConcurrentHashMap<String, Long> timeCache = new ConcurrentHashMap<>(); |
|
|
|
|
|
|
|
|
|
// 设备缓存，避免频繁查询数据库
|
|
|
|
|
private final ConcurrentHashMap<Long, CachedDeviceInfo> deviceCache = new ConcurrentHashMap<>(); |
|
|
|
|
|
|
|
|
|
// Device和Strategy实例缓存
|
|
|
|
|
private final ConcurrentHashMap<String, Device> deviceInstanceCache = new ConcurrentHashMap<>(); |
|
|
|
|
private final ConcurrentHashMap<String, DeviceStrategy> strategyInstanceCache = new ConcurrentHashMap<>(); |
|
|
|
|
|
|
|
|
|
// 缓存的设备信息
|
|
|
|
|
private static class CachedDeviceInfo { |
|
|
|
|
DeviceInstallEntity device; |
|
|
|
|
long timestamp; |
|
|
|
|
|
|
|
|
|
CachedDeviceInfo(DeviceInstallEntity device) { |
|
|
|
|
this.device = device; |
|
|
|
|
this.timestamp = System.currentTimeMillis(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
boolean isExpired() { |
|
|
|
|
return System.currentTimeMillis() - timestamp > TimeUnit.MINUTES.toMillis(DEVICE_CACHE_EXPIRE_MINUTES); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@PostConstruct |
|
|
|
|
public void init() { |
|
|
|
|
// 初始化线程池
|
|
|
|
|
deviceAnalysisExecutor = new ThreadPoolExecutor( |
|
|
|
|
CORE_POOL_SIZE, |
|
|
|
|
MAX_POOL_SIZE, |
|
|
|
|
KEEP_ALIVE_TIME, |
|
|
|
|
TimeUnit.SECONDS, |
|
|
|
|
new ArrayBlockingQueue<>(QUEUE_CAPACITY), |
|
|
|
|
new ThreadFactory() { |
|
|
|
|
private final AtomicInteger threadNumber = new AtomicInteger(1); |
|
|
|
|
@Override |
|
|
|
|
public Thread newThread(Runnable r) { |
|
|
|
|
return new Thread(r, "device-analysis-thread-" + threadNumber.getAndIncrement()); |
|
|
|
|
} |
|
|
|
|
}, |
|
|
|
|
new ThreadPoolExecutor.CallerRunsPolicy() // 改为CallerRuns，防止任务丢失
|
|
|
|
|
); |
|
|
|
|
|
|
|
|
|
// 启动缓存清理定时任务
|
|
|
|
|
ScheduledExecutorService cleanupExecutor = Executors.newSingleThreadScheduledExecutor(r -> |
|
|
|
|
new Thread(r, "cache-cleanup-thread")); |
|
|
|
|
cleanupExecutor.scheduleAtFixedRate(this::cleanupCaches, 5, 5, TimeUnit.MINUTES); |
|
|
|
|
|
|
|
|
|
log.info("EventsServiceImpl initialized, thread pool: core={}, max={}, queue={}", |
|
|
|
|
CORE_POOL_SIZE, MAX_POOL_SIZE, QUEUE_CAPACITY); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@PreDestroy |
|
|
|
|
public void destroy() { |
|
|
|
|
if (deviceAnalysisExecutor != null && !deviceAnalysisExecutor.isShutdown()) { |
|
|
|
|
deviceAnalysisExecutor.shutdown(); |
|
|
|
|
try { |
|
|
|
|
if (!deviceAnalysisExecutor.awaitTermination(30, TimeUnit.SECONDS)) { |
|
|
|
|
deviceAnalysisExecutor.shutdownNow(); |
|
|
|
|
} |
|
|
|
|
} catch (InterruptedException e) { |
|
|
|
|
deviceAnalysisExecutor.shutdownNow(); |
|
|
|
|
Thread.currentThread().interrupt(); |
|
|
|
|
} |
|
|
|
|
log.info("EventsServiceImpl destroyed, thread pool shutdown"); |
|
|
|
|
} |
|
|
|
|
clearAllCaches(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 清理过期缓存 |
|
|
|
|
*/ |
|
|
|
|
private void cleanupCaches() { |
|
|
|
|
try { |
|
|
|
|
long now = System.currentTimeMillis(); |
|
|
|
|
long expireTime = TimeUnit.MINUTES.toMillis(TIME_CACHE_EXPIRE_MINUTES); |
|
|
|
|
|
|
|
|
|
// 清理timeCache过期条目
|
|
|
|
|
timeCache.entrySet().removeIf(entry -> now - entry.getValue() > expireTime); |
|
|
|
|
|
|
|
|
|
// 清理deviceCache过期条目
|
|
|
|
|
deviceCache.entrySet().removeIf(entry -> entry.getValue().isExpired()); |
|
|
|
|
|
|
|
|
|
log.debug("缓存清理完成, timeCache.size={}, deviceCache.size={}", |
|
|
|
|
timeCache.size(), deviceCache.size()); |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("清理缓存失败", e); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 清空所有缓存 |
|
|
|
|
*/ |
|
|
|
|
private void clearAllCaches() { |
|
|
|
|
timeCache.clear(); |
|
|
|
|
deviceCache.clear(); |
|
|
|
|
deviceInstanceCache.clear(); |
|
|
|
|
strategyInstanceCache.clear(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ServiceActivator(inputChannel = ChannelName.EVENTS_UPLOAD_INBOUND) |
|
|
|
|
@Override |
|
|
|
|
public void handleInboundUpload(byte[] receiver, MessageHeaders headers) { |
|
|
|
|
@ -95,143 +220,431 @@ public class EventsServiceImpl implements IEventsService {
|
|
|
|
|
log.info("接收到控制指令下发=>{}", sendStr); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private void handleInboundData(byte[] receiver,String topic, String logMessage) { |
|
|
|
|
private void handleInboundData(byte[] receiver, String topic, String logMessage) { |
|
|
|
|
try { |
|
|
|
|
// 使用 TypeReference 确保泛型信息被保留
|
|
|
|
|
SanShiFengReceiver<SanShiFengDatas> datas = mapper.readValue(receiver, |
|
|
|
|
new TypeReference<SanShiFengReceiver<SanShiFengDatas>>() {}); |
|
|
|
|
log.info("主题：{}，类型：{}: ，数据：{}", topic, logMessage, datas.toString()); |
|
|
|
|
// log.info("主题：{}，类型：{}: ，数据：{}", topic, logMessage, datas.toString());
|
|
|
|
|
|
|
|
|
|
// 开始遍历 数据
|
|
|
|
|
String sn = datas.getSn(); |
|
|
|
|
String plcName = datas.getPlcName(); |
|
|
|
|
String projectName = datas.getProjectName(); |
|
|
|
|
String time = datas.getTime(); |
|
|
|
|
|
|
|
|
|
// 更新网关设备在线状态
|
|
|
|
|
gatewayManageService.updateGatewayManageOnlineBySn(sn, 0); |
|
|
|
|
|
|
|
|
|
// 获取网关对应的buildingId
|
|
|
|
|
String buildingId = gatewayManageService.queryBuildingIdBySn(sn); |
|
|
|
|
if (StringUtils.isBlank(buildingId)) { |
|
|
|
|
log.error("未找到对应的buildingId"); |
|
|
|
|
log.error("未找到对应的buildingId, SN: {}", sn); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
// 修复类型转换问题
|
|
|
|
|
|
|
|
|
|
// 获取数据列表
|
|
|
|
|
List<SanShiFengDatas> rawDataList = datas.getDatas(); |
|
|
|
|
if (rawDataList == null || rawDataList.isEmpty()) { |
|
|
|
|
log.warn("数据列表为空，SN: {}", sn); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
// rawDataList进行批量更新，100条数据进行批量处理
|
|
|
|
|
int batchSize = 100; |
|
|
|
|
for (int i = 0; i < rawDataList.size(); i += batchSize) { |
|
|
|
|
int endIndex = Math.min(i + batchSize, rawDataList.size()); |
|
|
|
|
|
|
|
|
|
// 批量更新collectionParams
|
|
|
|
|
processBatchUpdate(rawDataList, sn, plcName, projectName, time, buildingId); |
|
|
|
|
|
|
|
|
|
// 检查时间间隔并处理数据
|
|
|
|
|
if (shouldProcessData(sn, time)) { |
|
|
|
|
processDataList(rawDataList, sn, plcName, projectName, time, buildingId); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
} catch (IOException e) { |
|
|
|
|
log.error("处理数据时发生错误: ", e); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 批量更新collectionParams |
|
|
|
|
*/ |
|
|
|
|
private void processBatchUpdate(List<SanShiFengDatas> rawDataList, |
|
|
|
|
String sn, |
|
|
|
|
String plcName, |
|
|
|
|
String projectName, |
|
|
|
|
String time, |
|
|
|
|
String buildingId) { |
|
|
|
|
try { |
|
|
|
|
int size = rawDataList.size(); |
|
|
|
|
for (int i = 0; i < size; i += BATCH_SIZE) { |
|
|
|
|
int endIndex = Math.min(i + BATCH_SIZE, size); |
|
|
|
|
List<SanShiFengDatas> batch = rawDataList.subList(i, endIndex); |
|
|
|
|
collectionParamManageService.getBatchUpdateCollectionParams(batch, sn, plcName, projectName, time, buildingId); |
|
|
|
|
} |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("批量更新collectionParams失败: SN={}", sn, e); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 判断是否应该处理数据（基于时间间隔） |
|
|
|
|
*/ |
|
|
|
|
private boolean shouldProcessData(String sn, String time) { |
|
|
|
|
if (StringUtils.isBlank(sn) || StringUtils.isBlank(time)) { |
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
String cacheKey = sn + "_time"; |
|
|
|
|
Long lastTimestamp = timeCache.get(cacheKey); |
|
|
|
|
|
|
|
|
|
if (lastTimestamp == null) { |
|
|
|
|
timeCache.put(cacheKey, DateUtil.getTimeStamp(time)); |
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// // 先批量更新collectionParam
|
|
|
|
|
// rawDataList.parallelStream().forEach(rawData -> {
|
|
|
|
|
// try {
|
|
|
|
|
// processDataUpdateCpmItem(rawData, sn, plcName, projectName, time, buildingId);
|
|
|
|
|
// } catch (Exception e) {
|
|
|
|
|
// log.error("处理单个数据项失败: {}", rawData, e);
|
|
|
|
|
// }
|
|
|
|
|
// });
|
|
|
|
|
// 通过判断当前time跟上一个time相差30s才存储进入队列
|
|
|
|
|
if (caffeineCache.getIfPresent(sn+"_time") != null) { |
|
|
|
|
String lastTime = (String)caffeineCache.getIfPresent(sn+"_time"); |
|
|
|
|
// yyyy-MM-dd HH:mm:ss格式转为秒的时间戳
|
|
|
|
|
long lastTimeStamp = DateUtil.getTimeStamp(lastTime); |
|
|
|
|
try { |
|
|
|
|
long currentTimeStamp = DateUtil.getTimeStamp(time); |
|
|
|
|
// 判断时间间隔
|
|
|
|
|
if (!StringUtils.isBlank(lastTime) && Math.abs(currentTimeStamp -lastTimeStamp) >= 60000) { |
|
|
|
|
// 并行处理数据列表，主线程不阻塞
|
|
|
|
|
rawDataList.parallelStream().forEach(rawData -> { |
|
|
|
|
long timeDiff = Math.abs(currentTimeStamp - lastTimestamp); |
|
|
|
|
|
|
|
|
|
if (timeDiff >= TIME_INTERVAL_THRESHOLD_MS) { |
|
|
|
|
timeCache.put(cacheKey, currentTimeStamp); |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("计算时间间隔失败: SN={}, time={}", sn, time, e); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 处理数据列表（修复CountDownLatch泄露问题） |
|
|
|
|
*/ |
|
|
|
|
private void processDataList(List<SanShiFengDatas> rawDataList, |
|
|
|
|
String sn, |
|
|
|
|
String plcName, |
|
|
|
|
String projectName, |
|
|
|
|
String time, |
|
|
|
|
String buildingId) { |
|
|
|
|
if (rawDataList == null || rawDataList.isEmpty()) { |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 限制并发数，避免一次性创建过多任务
|
|
|
|
|
int size = rawDataList.size(); |
|
|
|
|
int maxConcurrent = Math.min(size, MAX_POOL_SIZE * 2); |
|
|
|
|
CountDownLatch latch = new CountDownLatch(size); |
|
|
|
|
AtomicInteger successCount = new AtomicInteger(0); |
|
|
|
|
AtomicInteger failCount = new AtomicInteger(0); |
|
|
|
|
|
|
|
|
|
// 使用信号量控制并发数
|
|
|
|
|
Semaphore semaphore = new Semaphore(maxConcurrent); |
|
|
|
|
|
|
|
|
|
for (int i = 0; i < size; i++) { |
|
|
|
|
final int index = i; |
|
|
|
|
final SanShiFengDatas data = rawDataList.get(i); |
|
|
|
|
|
|
|
|
|
// 提交任务
|
|
|
|
|
submitTaskWithFallback(() -> { |
|
|
|
|
try { |
|
|
|
|
processDataItem(rawData, sn, plcName, projectName, time, buildingId); |
|
|
|
|
semaphore.acquire(); |
|
|
|
|
try { |
|
|
|
|
processDataItem(data, sn, plcName, projectName, time, buildingId); |
|
|
|
|
successCount.incrementAndGet(); |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("处理单个数据项失败: {}", rawData, e); |
|
|
|
|
log.error("处理数据项失败: index={}", index, e); |
|
|
|
|
failCount.incrementAndGet(); |
|
|
|
|
} |
|
|
|
|
} catch (InterruptedException e) { |
|
|
|
|
Thread.currentThread().interrupt(); |
|
|
|
|
log.error("任务被中断: index={}", index, e); |
|
|
|
|
} finally { |
|
|
|
|
semaphore.release(); |
|
|
|
|
latch.countDown(); // 确保countDown一定会执行
|
|
|
|
|
} |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
caffeineCache.put(sn+"_time", time); |
|
|
|
|
|
|
|
|
|
// 等待所有任务完成
|
|
|
|
|
try { |
|
|
|
|
latch.await(30, TimeUnit.SECONDS); |
|
|
|
|
if (latch.getCount() > 0) { |
|
|
|
|
log.warn("部分任务未在指定时间内完成, remaining={}, success={}, fail={}", |
|
|
|
|
latch.getCount(), successCount.get(), failCount.get()); |
|
|
|
|
} |
|
|
|
|
} catch (IOException e) { |
|
|
|
|
log.error("处理数据时发生错误: ", e); |
|
|
|
|
} catch (InterruptedException e) { |
|
|
|
|
log.warn("等待数据处理完成被中断"); |
|
|
|
|
Thread.currentThread().interrupt(); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 提交任务，失败时直接在当前线程执行 |
|
|
|
|
*/ |
|
|
|
|
private void submitTaskWithFallback(Runnable task) { |
|
|
|
|
try { |
|
|
|
|
// 尝试提交到线程池
|
|
|
|
|
if (!deviceAnalysisExecutor.isShutdown()) { |
|
|
|
|
deviceAnalysisExecutor.submit(task); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
} catch (RejectedExecutionException | IllegalStateException e) { |
|
|
|
|
log.warn("任务提交失败，降级到当前线程执行"); |
|
|
|
|
} |
|
|
|
|
// 降级：在当前线程执行
|
|
|
|
|
task.run(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private void processDataUpdateCpmItem(SanShiFengDatas data, String sn, String plcName, String projectName, String time, String buildingId) { |
|
|
|
|
// 安全地转换对象
|
|
|
|
|
// SanShiFengDatas data = convertDataItem(rawData);
|
|
|
|
|
if (data == null) { |
|
|
|
|
log.warn("数据转换失败，跳过处理"); |
|
|
|
|
log.warn("数据为null，跳过处理"); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
// 不使用消息队列，查询属于哪种设备类型，然后通过策略进行数据解析
|
|
|
|
|
// log.info("设备SN：{}，PLC名称：{}，项目名称：{}，时间：{}，数据：{}", sn, plcName, projectName, time, data.toString());
|
|
|
|
|
// 获取点位参数名称
|
|
|
|
|
|
|
|
|
|
String name = data.getName(); |
|
|
|
|
// 获取点位值
|
|
|
|
|
BigDecimal value; |
|
|
|
|
try { |
|
|
|
|
value = data.getValue(); |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
if (StringUtils.isBlank(name)) { |
|
|
|
|
log.warn("点位名称为空，跳过处理"); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
BigDecimal value = data.getValue(); |
|
|
|
|
if (value == null) { |
|
|
|
|
value = BigDecimal.ZERO; |
|
|
|
|
} |
|
|
|
|
// 直接更新collectionParamManage参数值
|
|
|
|
|
|
|
|
|
|
try { |
|
|
|
|
collectionParamManageService.updateCPMByOtherName(name, value, time, buildingId); |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("更新collectionParamManage失败: name={}, value={}", name, value, e); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private void processDataItem(Object rawData, String sn, String plcName, String projectName, String time, String buildingId) { |
|
|
|
|
// 安全地转换对象
|
|
|
|
|
SanShiFengDatas data = convertDataItem(rawData); |
|
|
|
|
private void processDataItem(SanShiFengDatas data, String sn, String plcName, String projectName, String time, String buildingId) { |
|
|
|
|
if (data == null) { |
|
|
|
|
log.warn("数据转换失败，跳过处理"); |
|
|
|
|
log.warn("数据为null，跳过处理"); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
// 不使用消息队列，查询属于哪种设备类型，然后通过策略进行数据解析
|
|
|
|
|
// log.info("设备SN：{}，PLC名称：{}，项目名称：{}，时间：{}，数据：{}", sn, plcName, projectName, time, data.toString());
|
|
|
|
|
// 获取点位参数名称
|
|
|
|
|
|
|
|
|
|
// 获取点位参数名称和值
|
|
|
|
|
String name = data.getName(); |
|
|
|
|
// 获取点位值
|
|
|
|
|
BigDecimal value = new BigDecimal(0); |
|
|
|
|
if (StringUtils.isBlank(name)) { |
|
|
|
|
log.warn("点位名称为空，跳过处理"); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
BigDecimal value = data.getValue(); |
|
|
|
|
if (value == null) { |
|
|
|
|
value = BigDecimal.ZERO; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
try { |
|
|
|
|
value = new BigDecimal(String.valueOf(data.getValue())); |
|
|
|
|
// 获取collectionParams缓存，使用带过期时间的本地缓存
|
|
|
|
|
List<CollectionParamsManageEntity> collectionParams = getCollectionParamsCache(); |
|
|
|
|
|
|
|
|
|
if (collectionParams == null || collectionParams.isEmpty()) { |
|
|
|
|
log.debug("collectionParams缓存为空"); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 查找匹配的参数实体
|
|
|
|
|
CollectionParamsManageEntity collectionParamsManageEntity = findMatchingCollectionParams(collectionParams, name, buildingId); |
|
|
|
|
|
|
|
|
|
if (collectionParamsManageEntity == null) { |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 检查参数类型，过滤不需要处理的类型
|
|
|
|
|
if (!shouldProcessParamType(collectionParamsManageEntity.getParamTypeId())) { |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 查询设备信息并处理
|
|
|
|
|
processDeviceData(collectionParamsManageEntity, time, value); |
|
|
|
|
|
|
|
|
|
} catch (Exception e) { |
|
|
|
|
value = BigDecimal.ZERO; |
|
|
|
|
log.error("处理数据项失败: name={}, value={}", name, value, e); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 获取collectionParams缓存 |
|
|
|
|
*/ |
|
|
|
|
private List<CollectionParamsManageEntity> getCollectionParamsCache() { |
|
|
|
|
try { |
|
|
|
|
// 优先使用caffeine缓存
|
|
|
|
|
List<CollectionParamsManageEntity> cachedParams = (List<CollectionParamsManageEntity>) caffeineCache.getIfPresent("collectionParams"); |
|
|
|
|
|
|
|
|
|
if (cachedParams != null) { |
|
|
|
|
return cachedParams; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 从数据库加载
|
|
|
|
|
List<CollectionParamsManageEntity> collectionParams = collectionParamManageService.selectAllCPMList(); |
|
|
|
|
|
|
|
|
|
if (collectionParams != null) { |
|
|
|
|
// 放入caffeine缓存
|
|
|
|
|
caffeineCache.put("collectionParams", collectionParams); |
|
|
|
|
log.info("collectionParams已加载到缓存，共{}条记录", collectionParams.size()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
return collectionParams; |
|
|
|
|
|
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("获取collectionParams缓存失败", e); |
|
|
|
|
return null; |
|
|
|
|
} |
|
|
|
|
// 直接更新collectionParamManage参数值
|
|
|
|
|
//collectionParamManageService.updateCPMByOtherName(name, value, time, buildingId);
|
|
|
|
|
// 查询device_install表，走之前的逻辑
|
|
|
|
|
CollectionParamsManageEntity collectionParamsManageEntity = collectionParamManageService.selectDeviceInstallByOtherName(name, buildingId); |
|
|
|
|
if (null != collectionParamsManageEntity |
|
|
|
|
&& collectionParamsManageEntity.getDeviceInstallId() != null |
|
|
|
|
&& collectionParamsManageEntity.getParamTypeId() != 0 |
|
|
|
|
&& collectionParamsManageEntity.getParamTypeId() != 4 |
|
|
|
|
&& collectionParamsManageEntity.getParamTypeId() != 15 |
|
|
|
|
&& collectionParamsManageEntity.getParamTypeId() != 16 |
|
|
|
|
&& collectionParamsManageEntity.getParamTypeId() != 17 |
|
|
|
|
&& collectionParamsManageEntity.getParamTypeId() != 18 |
|
|
|
|
&& collectionParamsManageEntity.getParamTypeId() != 19 |
|
|
|
|
&& collectionParamsManageEntity.getParamTypeId() != 21 |
|
|
|
|
&& collectionParamsManageEntity.getParamTypeId() != 22 |
|
|
|
|
&& collectionParamsManageEntity.getParamTypeId() != 23 |
|
|
|
|
&& collectionParamsManageEntity.getParamTypeId() != 24 |
|
|
|
|
&& collectionParamsManageEntity.getParamTypeId() != 3 // 通过运行状态点判断故障点
|
|
|
|
|
) { |
|
|
|
|
DeviceInstallEntity deviceInstallEntity = deviceInstallService.selectDeviceById(collectionParamsManageEntity.getDeviceInstallId()); |
|
|
|
|
if (deviceInstallEntity != null) { |
|
|
|
|
// 开始走策略判断
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 查找匹配的collectionParams |
|
|
|
|
*/ |
|
|
|
|
private CollectionParamsManageEntity findMatchingCollectionParams(List<CollectionParamsManageEntity> collectionParams, |
|
|
|
|
String name, |
|
|
|
|
String buildingId) { |
|
|
|
|
if (collectionParams == null || StringUtils.isBlank(name) || StringUtils.isBlank(buildingId)) { |
|
|
|
|
return null; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
try { |
|
|
|
|
for (CollectionParamsManageEntity val : collectionParams) { |
|
|
|
|
if (val == null) { |
|
|
|
|
continue; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
String otherName = val.getOtherName(); |
|
|
|
|
if (otherName != null && otherName.trim().equals(name.trim())) { |
|
|
|
|
Object bidObj = val.getBuildingId(); |
|
|
|
|
if (bidObj != null && bidObj.toString().equals(buildingId)) { |
|
|
|
|
return val; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("查找collectionParams失败: name={}, buildingId={}", name, buildingId, e); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
return null; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 判断参数类型是否需要处理 |
|
|
|
|
*/ |
|
|
|
|
private boolean shouldProcessParamType(int paramTypeId) { |
|
|
|
|
// 不需要处理的参数类型
|
|
|
|
|
int[] excludedTypes = {0, 3, 4, 15, 16, 17, 18, 19, 21, 22, 23, 24}; |
|
|
|
|
|
|
|
|
|
for (int excluded : excludedTypes) { |
|
|
|
|
if (paramTypeId == excluded) { |
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 处理设备数据（优化：使用缓存减少数据库查询和对象创建） |
|
|
|
|
*/ |
|
|
|
|
private void processDeviceData(CollectionParamsManageEntity collectionParamsManageEntity, |
|
|
|
|
String time, |
|
|
|
|
BigDecimal value) { |
|
|
|
|
if (collectionParamsManageEntity == null || |
|
|
|
|
collectionParamsManageEntity.getDeviceInstallId() == null) { |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
Long deviceInstallId = collectionParamsManageEntity.getDeviceInstallId(); |
|
|
|
|
|
|
|
|
|
try { |
|
|
|
|
// 从缓存获取设备信息
|
|
|
|
|
DeviceInstallEntity deviceInstallEntity = getDeviceFromCache(deviceInstallId); |
|
|
|
|
|
|
|
|
|
if (deviceInstallEntity == null) { |
|
|
|
|
log.warn("设备信息不存在: deviceInstallId={}", deviceInstallId); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
String deviceType = deviceInstallEntity.getDeviceType(); |
|
|
|
|
Device device = DeviceFactory.createDevice(deviceType); |
|
|
|
|
DeviceStrategy strategy = DeviceStrategyFactory.createStrategy(deviceType); |
|
|
|
|
if (strategy != null) { |
|
|
|
|
if (StringUtils.isBlank(deviceType)) { |
|
|
|
|
log.warn("设备类型为空: deviceInstallId={}", deviceInstallEntity.getId()); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 从缓存获取Device和Strategy实例
|
|
|
|
|
Device device = getDeviceInstance(deviceType); |
|
|
|
|
DeviceStrategy strategy = getStrategyInstance(deviceType); |
|
|
|
|
|
|
|
|
|
if (device == null || strategy == null) { |
|
|
|
|
log.warn("创建设备或策略失败: deviceType={}", deviceType); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 执行分析
|
|
|
|
|
device.setStrategy(strategy); |
|
|
|
|
device.analysisMQTTReceiveData(time, deviceInstallEntity.getDeviceAddr(), value.toPlainString(), Constant.READ, deviceInstallEntity, collectionParamsManageEntity); |
|
|
|
|
device.analysisMQTTReceiveData( |
|
|
|
|
time, |
|
|
|
|
deviceInstallEntity.getDeviceAddr(), |
|
|
|
|
value.toPlainString(), |
|
|
|
|
Constant.READ, |
|
|
|
|
deviceInstallEntity, |
|
|
|
|
collectionParamsManageEntity); |
|
|
|
|
|
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("处理设备数据失败: time={}, value={}", time, value, e); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 从缓存获取设备信息 |
|
|
|
|
*/ |
|
|
|
|
private DeviceInstallEntity getDeviceFromCache(Long deviceInstallId) { |
|
|
|
|
if (deviceInstallId == null) { |
|
|
|
|
return null; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 先从缓存获取
|
|
|
|
|
CachedDeviceInfo cachedInfo = deviceCache.get(deviceInstallId); |
|
|
|
|
|
|
|
|
|
if (cachedInfo != null && !cachedInfo.isExpired()) { |
|
|
|
|
return cachedInfo.device; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 缓存未命中或已过期，从数据库加载
|
|
|
|
|
try { |
|
|
|
|
DeviceInstallEntity device = deviceInstallService.selectDeviceById(deviceInstallId); |
|
|
|
|
if (device != null) { |
|
|
|
|
deviceCache.put(deviceInstallId, new CachedDeviceInfo(device)); |
|
|
|
|
} |
|
|
|
|
return device; |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("查询设备信息失败: deviceInstallId={}", deviceInstallId, e); |
|
|
|
|
return null; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 获取Device实例（使用缓存） |
|
|
|
|
*/ |
|
|
|
|
private Device getDeviceInstance(String deviceType) { |
|
|
|
|
return deviceInstanceCache.computeIfAbsent(deviceType, type -> { |
|
|
|
|
try { |
|
|
|
|
return DeviceFactory.createDevice(type); |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("创建Device实例失败: deviceType={}", type, e); |
|
|
|
|
return null; |
|
|
|
|
} |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 获取DeviceStrategy实例（使用缓存） |
|
|
|
|
*/ |
|
|
|
|
private DeviceStrategy getStrategyInstance(String deviceType) { |
|
|
|
|
return strategyInstanceCache.computeIfAbsent(deviceType, type -> { |
|
|
|
|
try { |
|
|
|
|
return DeviceStrategyFactory.createStrategy(type); |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("创建DeviceStrategy实例失败: deviceType={}", type, e); |
|
|
|
|
return null; |
|
|
|
|
} |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private SanShiFengDatas convertDataItem(Object rawData) { |
|
|
|
|
@ -239,24 +652,29 @@ public class EventsServiceImpl implements IEventsService {
|
|
|
|
|
return null; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
SanShiFengDatas data = new SanShiFengDatas(); |
|
|
|
|
try { |
|
|
|
|
if (rawData instanceof SanShiFengDatas) { |
|
|
|
|
data = (SanShiFengDatas) rawData; |
|
|
|
|
return (SanShiFengDatas) rawData; |
|
|
|
|
} else if (rawData instanceof HashMap) { |
|
|
|
|
JSONObject jsonObject = new JSONObject((HashMap<?, ?>) rawData); |
|
|
|
|
data = jsonObject.to(SanShiFengDatas.class); |
|
|
|
|
return jsonObject.to(SanShiFengDatas.class); |
|
|
|
|
} else { |
|
|
|
|
log.warn("不支持的数据类型: {}", rawData.getClass().getName()); |
|
|
|
|
return null; |
|
|
|
|
} |
|
|
|
|
} catch (Exception e) { |
|
|
|
|
log.error("数据转换异常", e); |
|
|
|
|
data.setName(getJsonValueAsString(rawData, "name")); |
|
|
|
|
data.setValue(new BigDecimal("-1")); |
|
|
|
|
} |
|
|
|
|
// 尝试恢复至少name字段
|
|
|
|
|
String name = getJsonValueAsString(rawData, "name"); |
|
|
|
|
if (!StringUtils.isBlank(name)) { |
|
|
|
|
SanShiFengDatas data = new SanShiFengDatas(); |
|
|
|
|
data.setName(name); |
|
|
|
|
data.setValue(BigDecimal.ZERO); |
|
|
|
|
return data; |
|
|
|
|
} |
|
|
|
|
return null; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
* 从原始数据对象中获取指定键的字符串值 |
|
|
|
|
@ -279,7 +697,6 @@ public class EventsServiceImpl implements IEventsService {
|
|
|
|
|
JSONObject jsonObject = (JSONObject) rawData; |
|
|
|
|
return jsonObject.getString(key); |
|
|
|
|
} else { |
|
|
|
|
// 如果是其他类型，尝试使用反射或通用方式获取
|
|
|
|
|
log.warn("不支持的数据类型: {}", rawData.getClass().getName()); |
|
|
|
|
return null; |
|
|
|
|
} |
|
|
|
|
@ -289,5 +706,4 @@ public class EventsServiceImpl implements IEventsService {
|
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
} |
|
|
|
|
|
