diff --git a/iot-dao/iot-data-serviceImpl-rdb/src/main/java/cc/iotkit/data/dao/DiccidRecordRepository.java b/iot-dao/iot-data-serviceImpl-rdb/src/main/java/cc/iotkit/data/dao/DiccidRecordRepository.java index 8ebd9c5..1e6aed9 100644 --- a/iot-dao/iot-data-serviceImpl-rdb/src/main/java/cc/iotkit/data/dao/DiccidRecordRepository.java +++ b/iot-dao/iot-data-serviceImpl-rdb/src/main/java/cc/iotkit/data/dao/DiccidRecordRepository.java @@ -38,5 +38,5 @@ public interface DiccidRecordRepository extends JpaRepository findAll(); - TbDIccidRecordDO findByImei(String imei); + TbDIccidRecordDO findFirstByImei(String imei); } diff --git a/iot-dao/iot-data-serviceImpl-rdb/src/main/java/cc/iotkit/data/service/DIccidRecordDataImpl.java b/iot-dao/iot-data-serviceImpl-rdb/src/main/java/cc/iotkit/data/service/DIccidRecordDataImpl.java index 3e738fd..b39f466 100644 --- a/iot-dao/iot-data-serviceImpl-rdb/src/main/java/cc/iotkit/data/service/DIccidRecordDataImpl.java +++ b/iot-dao/iot-data-serviceImpl-rdb/src/main/java/cc/iotkit/data/service/DIccidRecordDataImpl.java @@ -77,7 +77,7 @@ public class DIccidRecordDataImpl implements IDIccidRecordData, IJPACommData imageCache = new ConcurrentHashMap<>(); - // 线程池用于并发处理图片 private static final ExecutorService imageProcessingPool = Executors.newFixedThreadPool(4); @@ -87,7 +87,7 @@ public class PDFGenerator { {"报警器编号", entity.getDeviceName()}, {"切断阀编号", entity.getShutValueNumber()}, {"燃气表号", entity.getGasMeterNumber()}, - {"备注", entity.getBuildingUnit()} + {"备注", entity.getRemarks()} }; drawTable(cs, font, TABLE_MARGIN, currentY, baseData, COLUMN_WIDTHS); currentY -= (ROW_HEIGHT * baseData.length) + 20f; @@ -396,25 +396,34 @@ public class PDFGenerator { return PDType0Font.load(document, new File("/ttf/NotoSansCJK-Regular.ttf")); } - // 缓存已处理的图片以避免重复下载和处理 - private static final ConcurrentHashMap imageByteCache = new ConcurrentHashMap<>(); - + // LRU缓存图片字节，最多50张，30分钟未访问自动淘汰 + private static final Cache imageByteCache = CacheBuilder.newBuilder() + .maximumSize(50) + .expireAfterAccess(30, TimeUnit.MINUTES) + .build(); // 预加载图片以避免重复下载 private static PDImageXObject loadImageFromUrl(PDDocument document, String url) throws IOException { - byte[] imageBytes = imageByteCache.computeIfAbsent(url, key -> { - try { - return new URL(key).openStream().readAllBytes(); - } catch (IOException e) { - return null; // 或者抛出异常 + try { + byte[] imageBytes = imageByteCache.get(url, () -> { + try { + return new URL(url).openStream().readAllBytes(); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + + if (imageBytes == null) { + return null; } - }); - if (imageBytes == null) { - return null; + return PDImageXObject.createFromByteArray(document, imageBytes, ""); + } catch (Exception e) { + if (e.getCause() instanceof IOException) { + throw (IOException) e.getCause(); + } + throw new IOException("Failed to load image: " + url, e); } - - return PDImageXObject.createFromByteArray(document, imageBytes, ""); } diff --git a/iot-module/iot-manager/src/main/java/cc/iotkit/manager/dto/vo/deviceinfo/DeviceInstallInfoExpordVo.java b/iot-module/iot-manager/src/main/java/cc/iotkit/manager/dto/vo/deviceinfo/DeviceInstallInfoExpordVo.java index 057cf5e..7595bb3 100644 --- a/iot-module/iot-manager/src/main/java/cc/iotkit/manager/dto/vo/deviceinfo/DeviceInstallInfoExpordVo.java +++ b/iot-module/iot-manager/src/main/java/cc/iotkit/manager/dto/vo/deviceinfo/DeviceInstallInfoExpordVo.java @@ -58,9 +58,12 @@ public class DeviceInstallInfoExpordVo implements Serializable { // @ApiModelProperty(value = "用户电话") @ExcelProperty(value = "电话") private String userIpone; - // @ApiModelProperty(value = "切断阀编号") - @ExcelProperty(value = "地址") - private String site; + @ExcelProperty(value = "小区名称") + private String communityName; + @ExcelProperty(value = "楼栋单元号") + private String buildingUnit; + @ExcelProperty(value = "房间号") + private String roomNo; @ExcelProperty(value = "切断阀编号") private String shutValueNumber; @ExcelProperty(value = "报警器编号") @@ -79,8 +82,6 @@ public class DeviceInstallInfoExpordVo implements Serializable { private String manufacturer; @ExcelProperty(value = "安装情况") private String position; - /* // @ApiModelProperty(value = "备注") + @ExcelProperty(value = "备注") private String remarks; - - private Integer state;*/ } diff --git a/iot-module/iot-manager/src/main/java/cc/iotkit/manager/dto/vo/deviceinfo/DeviceInstallInfoVo.java b/iot-module/iot-manager/src/main/java/cc/iotkit/manager/dto/vo/deviceinfo/DeviceInstallInfoVo.java index e12d8f0..68ab9cc 100644 --- a/iot-module/iot-manager/src/main/java/cc/iotkit/manager/dto/vo/deviceinfo/DeviceInstallInfoVo.java +++ b/iot-module/iot-manager/src/main/java/cc/iotkit/manager/dto/vo/deviceinfo/DeviceInstallInfoVo.java @@ -61,9 +61,12 @@ public class DeviceInstallInfoVo implements Serializable { // @ApiModelProperty(value = "用户电话") @ExcelProperty(value = "电话") private String userIpone; - // @ApiModelProperty(value = "切断阀编号") - @ExcelProperty(value = "地址") - private String site; + @ExcelProperty(value = "小区名称") + private String communityName; + @ExcelProperty(value = "楼栋单元号") + private String buildingUnit; + @ExcelProperty(value = "房间号") + private String roomNo; @ExcelProperty(value = "切断阀编号") private String shutValueNumber; @ExcelProperty(value = "报警器编号") @@ -82,8 +85,6 @@ public class DeviceInstallInfoVo implements Serializable { private String manufacturer; @ExcelProperty(value = "安装情况") private String position; - /* // @ApiModelProperty(value = "备注") + @ExcelProperty(value = "备注") private String remarks; - - private Integer state;*/ } diff --git a/iot-module/iot-manager/src/main/java/cc/iotkit/manager/service/impl/DeviceInstallInfoServiceImpl.java b/iot-module/iot-manager/src/main/java/cc/iotkit/manager/service/impl/DeviceInstallInfoServiceImpl.java index 5a1c764..b5bb82a 100644 --- a/iot-module/iot-manager/src/main/java/cc/iotkit/manager/service/impl/DeviceInstallInfoServiceImpl.java +++ b/iot-module/iot-manager/src/main/java/cc/iotkit/manager/service/impl/DeviceInstallInfoServiceImpl.java @@ -225,7 +225,10 @@ public class DeviceInstallInfoServiceImpl implements IDeviceInstallInfoService { deviceInstallInfoVo.setXuhao(i); deviceInstallInfoVo.setShutValueNumber(deviceInfo.getShutValueNumber()); deviceInstallInfoVo.setDeviceName(deviceInfo.getDeviceName()); - deviceInstallInfoVo.setSite(tbDeviceInfo.getCommunityName() + tbDeviceInfo.getBuildingUnit() + tbDeviceInfo.getRoomNo()); + deviceInstallInfoVo.setCommunityName(tbDeviceInfo.getCommunityName()); + deviceInstallInfoVo.setBuildingUnit(tbDeviceInfo.getBuildingUnit()); + deviceInstallInfoVo.setRoomNo(tbDeviceInfo.getRoomNo()); + deviceInstallInfoVo.setRemarks(tbDeviceInfo.getRemarks()); deviceInstallInfoVo.setManufacturer("天津费加罗"); deviceInstallInfoVo.setPosition("成功"); diff --git a/iot-module/iot-rule-engine/src/main/java/cc/iotkit/ruleengine/handler/RuleDeviceConsumer.java b/iot-module/iot-rule-engine/src/main/java/cc/iotkit/ruleengine/handler/RuleDeviceConsumer.java index 7c5676a..85f18dc 100644 --- a/iot-module/iot-rule-engine/src/main/java/cc/iotkit/ruleengine/handler/RuleDeviceConsumer.java +++ b/iot-module/iot-rule-engine/src/main/java/cc/iotkit/ruleengine/handler/RuleDeviceConsumer.java @@ -25,69 +25,65 @@ package cc.iotkit.ruleengine.handler; import cc.iotkit.common.constant.Constants; import cc.iotkit.common.thing.ThingModelMessage; -import cc.iotkit.common.utils.JsonUtils; -import cc.iotkit.common.utils.ThreadUtil; import cc.iotkit.mq.ConsumerHandler; import cc.iotkit.mq.MqConsumer; -import com.google.common.util.concurrent.ThreadFactoryBuilder; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.BeansException; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import javax.annotation.PreDestroy; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; @Slf4j public class RuleDeviceConsumer implements ConsumerHandler, ApplicationContextAware { private final List handlers = new ArrayList<>(); - private ScheduledThreadPoolExecutor messageHandlerPool; + private ThreadPoolExecutor messageHandlerPool; + + /** 连续失败计数器 */ + private final AtomicInteger consecutiveFailures = new AtomicInteger(0); + /** 熔断打开截止时间戳，0 表示关闭 */ + private volatile long circuitOpenUntil = 0; + + private static final int FAILURE_THRESHOLD = 10; + private static final long CIRCUIT_COOLDOWN_MS = 30_000; + private static final int QUEUE_CAPACITY = 500; - //private ThreadPoolExecutor messageHandlerPool; @SneakyThrows public RuleDeviceConsumer(MqConsumer consumer) { consumer.consume(Constants.THING_MODEL_MESSAGE_TOPIC, this); } - /* @Override + + @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { Map handlerMap = applicationContext.getBeansOfType(DeviceMessageHandler.class); - - // 动态计算核心线程数 - int corePoolSize = Math.max(4, Runtime.getRuntime().availableProcessors()); - // corePoolSize = corePoolSize > handlerMap.size()?corePoolSize:handlerMap.size(); - int maxPoolSize = corePoolSize * 2; - + int coreSize = handlerMap.size(); messageHandlerPool = new ThreadPoolExecutor( - corePoolSize, - maxPoolSize, + coreSize, + coreSize * 2, 60L, TimeUnit.SECONDS, - new LinkedBlockingQueue<>(10), // 限制队列大小 - new ThreadFactoryBuilder().setNameFormat("msg-handler-%d").build(), - new ThreadPoolExecutor.CallerRunsPolicy() // 饱和策略 + new LinkedBlockingQueue<>(QUEUE_CAPACITY), + new ThreadPoolExecutor.CallerRunsPolicy() ); - - this.handlers.addAll(handlerMap.values()); - }*/ - @Override - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - Map handlerMap = applicationContext.getBeansOfType(DeviceMessageHandler.class); - messageHandlerPool = ThreadUtil.newScheduled(handlerMap.size() * 2, "messageHandler"); this.handlers.addAll(handlerMap.values()); } @SneakyThrows @Override public void handler(ThingModelMessage msg) { + if (circuitOpenUntil > 0 && circuitOpenUntil > System.currentTimeMillis()) { + return; + } + log.info("received thing model message:{}", msg); try { for (DeviceMessageHandler handler : this.handlers) { @@ -97,8 +93,17 @@ public class RuleDeviceConsumer implements ConsumerHandler, A msg.setData(new HashMap<>()); } handler.handle(msg); + // 成功 → 重置熔断 + consecutiveFailures.set(0); + circuitOpenUntil = 0; } catch (Throwable e) { log.error("handler message error", e); + int failures = consecutiveFailures.incrementAndGet(); + if (failures >= FAILURE_THRESHOLD) { + circuitOpenUntil = System.currentTimeMillis() + CIRCUIT_COOLDOWN_MS; + log.error("Circuit breaker OPEN for {}s, {} consecutive failures", + CIRCUIT_COOLDOWN_MS / 1000, failures); + } } }); } @@ -107,4 +112,19 @@ public class RuleDeviceConsumer implements ConsumerHandler, A } } + @PreDestroy + public void shutdown() { + if (messageHandlerPool != null) { + messageHandlerPool.shutdown(); + try { + if (!messageHandlerPool.awaitTermination(10, TimeUnit.SECONDS)) { + messageHandlerPool.shutdownNow(); + } + } catch (InterruptedException e) { + messageHandlerPool.shutdownNow(); + Thread.currentThread().interrupt(); + } + } + } + }