diff --git a/intc-admin/src/main/resources/application.yml b/intc-admin/src/main/resources/application.yml index 860b173..d25d527 100644 --- a/intc-admin/src/main/resources/application.yml +++ b/intc-admin/src/main/resources/application.yml @@ -138,6 +138,7 @@ tenant: - sys_user_role - sys_client - sys_oss_config + - iot_device_status - flow_spel # MyBatisPlus配置 diff --git a/intc-modules/intc-fishery/src/main/java/com/intc/fishery/domain/vo/DeviceVo.java b/intc-modules/intc-fishery/src/main/java/com/intc/fishery/domain/vo/DeviceVo.java index fc14128..344aa33 100644 --- a/intc-modules/intc-fishery/src/main/java/com/intc/fishery/domain/vo/DeviceVo.java +++ b/intc-modules/intc-fishery/src/main/java/com/intc/fishery/domain/vo/DeviceVo.java @@ -336,12 +336,17 @@ public class DeviceVo implements Serializable { @ExcelProperty(value = "用户名") private String userName; - /** + /** * 塘口名称 */ @ExcelProperty(value = "塘口名称") private String pondName; + /** + * 设备在线状态（online/offline） + */ + private String onlineStatus; + /** * 是否过期（0-未过期，1-已过期） */ diff --git a/intc-modules/intc-fishery/src/main/java/com/intc/fishery/service/impl/DeviceServiceImpl.java b/intc-modules/intc-fishery/src/main/java/com/intc/fishery/service/impl/DeviceServiceImpl.java index 80e1319..59d711c 100644 --- a/intc-modules/intc-fishery/src/main/java/com/intc/fishery/service/impl/DeviceServiceImpl.java +++ b/intc-modules/intc-fishery/src/main/java/com/intc/fishery/service/impl/DeviceServiceImpl.java @@ -5,6 +5,7 @@ import java.util.stream.Collectors; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import com.intc.common.core.config.properties.DeviceTypeProperties; import com.intc.common.core.utils.MapstructUtils; import com.intc.common.core.utils.StringUtils; import com.intc.common.mybatis.core.page.PageQuery; @@ -60,11 +61,22 @@ public class DeviceServiceImpl implements IDeviceService { private final DeviceSwitchMapper deviceSwitchMapper; private final LinkedCtrlMapper linkedCtrlMapper; private final TimingCtrlMapper timingCtrlMapper; + private final DeviceTypeProperties deviceTypeProperties; /** Redis 离线等待标记前缀（与 DeviceDataHandler 保持一致） */ private static final String DEVICE_OFFLINE_WAIT_KEY_PREFIX = "device:offline:wait:"; /** Redis 溶解氧探头离线等待标记前缀 */ private static final String DETECTOR_OXY_OFFLINE_WAIT_KEY_PREFIX = "device:detector:oxy:offline:wait:"; + /** 太阳能网控设备类型 */ + private static final Integer DEVICE_TYPE_SOLAR_CONTROLLER = 3; + /** 设备在线状态表别名 */ + private static final String DEVICE_STATUS_ALIAS = "ids"; + /** 设备在线状态表状态字段 */ + private static final String DEVICE_STATUS_COLUMN = DEVICE_STATUS_ALIAS + ".status"; + /** 设备在线状态展示字段 */ + private static final String DEVICE_STATUS_DISPLAY_COLUMN = "LOWER(COALESCE(" + DEVICE_STATUS_COLUMN + ", 'offline'))"; + /** 设备在线状态表状态字段（小写） */ + private static final String DEVICE_STATUS_LOWER_COLUMN = "LOWER(" + DEVICE_STATUS_COLUMN + ")"; /** * 查询设备管理 @@ -115,6 +127,7 @@ public class DeviceServiceImpl implements IDeviceService { private MPJLambdaWrapper buildQueryWrapper(DeviceBo bo) { Map params = bo.getParams(); MPJLambdaWrapper wrapper = new MPJLambdaWrapper<>(); + boolean solarQuery = DEVICE_TYPE_SOLAR_CONTROLLER.equals(bo.getDeviceType()); // Select main table fields wrapper.selectAll(Device.class); @@ -127,6 +140,12 @@ public class DeviceServiceImpl implements IDeviceService { wrapper.leftJoin(AquUser.class, AquUser::getId, Device::getUserId); // Left join with pond table wrapper.leftJoin(Pond.class, Pond::getId, Device::getPondId); + if (solarQuery) { + wrapper.selectAs(DEVICE_STATUS_DISPLAY_COLUMN, DeviceVo::getOnlineStatus) + .leftJoin(buildSolarStatusJoinSql()); + applySolarOnlineStatusFilter(wrapper, params); + wrapper.orderByAsc("CASE WHEN " + DEVICE_STATUS_LOWER_COLUMN + " = 'online' THEN 0 ELSE 1 END"); + } // 处理排序逼辑：如果expiredFlag为1，按过期时间倒序，否则按创建时间倒序 String expiredFlag = params != null ? (String) params.get("expiredFlag") : null; @@ -154,6 +173,33 @@ public class DeviceServiceImpl implements IDeviceService { return wrapper; } + private String buildSolarStatusJoinSql() { + String sql = "iot_device_status " + DEVICE_STATUS_ALIAS + " ON " + + DEVICE_STATUS_ALIAS + ".device_name = t.serial_num"; + String productKey = deviceTypeProperties.getSolarController(); + if (StringUtils.isNotBlank(productKey)) { + sql += " AND " + DEVICE_STATUS_ALIAS + ".product_key = '" + productKey.replace("'", "''") + "'"; + } + return sql; + } + + /** + * 太阳能网控在线状态查询。 + */ + private void applySolarOnlineStatusFilter(MPJLambdaWrapper wrapper, Map params) { + if (params == null || params.isEmpty()) { + return; + } + String onlineStatus = params.get("onlineStatus") != null ? params.get("onlineStatus").toString() : null; + if ("online".equalsIgnoreCase(onlineStatus)) { + wrapper.apply(DEVICE_STATUS_LOWER_COLUMN + " = {0}", "online"); + } else if ("offline".equalsIgnoreCase(onlineStatus)) { + wrapper.and(w -> w.isNull(DEVICE_STATUS_COLUMN) + .or() + .apply(DEVICE_STATUS_LOWER_COLUMN + " <> {0}", "online")); + } + } + /** * 处理额外的查询参数 * diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java index 6718ed3..d62e65a 100644 --- a/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java +++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java @@ -63,6 +63,7 @@ import java.util.concurrent.TimeUnit; public class DeviceDataHandler { public static final String PENDING_SOLAR_PROPERTY_KEY_PREFIX = "device:pending:solar:property:"; + private static final String PENDING_SOLAR_PROPERTY_FLUSH_LOCK_PREFIX = "device:pending:solar:property:flush:"; private static final int DEVICE_TYPE_SOLAR_CONTROLLER = 3; /** @@ -1011,8 +1012,23 @@ public class DeviceDataHandler { deviceName, device.getId(), device.getDeviceType()); return; } + if (StrUtil.isBlank(device.getSerialNum())) { + log.debug("[太阳能属性] 上线设备编号为空，跳过补发: deviceName={}, deviceId={}, iotId={}", + deviceName, device.getId(), device.getIotId()); + return; + } + + String flushLockKey = PENDING_SOLAR_PROPERTY_FLUSH_LOCK_PREFIX + device.getSerialNum(); + boolean locked = false; try { + locked = RedisUtils.setObjectIfAbsent(flushLockKey, "1", Duration.ofSeconds(30)); + if (!locked) { + log.debug("[太阳能属性] 已有补发任务执行中，跳过重复补发: deviceName={}, deviceId={}", + deviceName, device.getId()); + return; + } + PendingSolarProperty pendingProperty = findPendingSolarProperty(device); if (pendingProperty == null) { logNoPendingSolarProperty(trigger, deviceName, device); @@ -1047,6 +1063,15 @@ public class DeviceDataHandler { trigger, deviceName, pendingProperty.redisKey, pendingProperty.deviceId, pendingProperty.properties); } catch (Exception e) { log.error("[太阳能属性] {}补发异常: deviceName={}, err={}", trigger, deviceName, e.getMessage(), e); + } finally { + if (locked) { + try { + RedisUtils.deleteObject(flushLockKey); + } catch (Exception e) { + log.warn("[太阳能属性] 释放补发锁失败: deviceName={}, lockKey={}, err={}", + deviceName, flushLockKey, e.getMessage()); + } + } } } diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/AmqpMessageHandlerImpl.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/AmqpMessageHandlerImpl.java index c0f2aa6..17d829d 100644 --- a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/AmqpMessageHandlerImpl.java +++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/AmqpMessageHandlerImpl.java @@ -5,11 +5,15 @@ import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import com.intc.iot.handler.DeviceDataHandler; import com.intc.iot.service.AmqpMessageHandler; +import com.intc.iot.service.DeviceStatusService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; +import java.util.HashMap; +import java.util.Map; + /** * AMQP 消息处理器实现类 * @@ -29,6 +33,9 @@ public class AmqpMessageHandlerImpl implements AmqpMessageHandler { @Autowired(required = false) private DeviceDataHandler deviceDataHandler; + @Autowired(required = false) + private DeviceStatusService deviceStatusService; + @Override public void handleMessage(String message) { try { @@ -218,17 +225,11 @@ public class AmqpMessageHandlerImpl implements AmqpMessageHandler { private void handleStatusMessage(JSONObject json) { String deviceName = json.getStr("deviceName"); String status = json.getStr("status"); - if (status != null) { - status = status.trim().toLowerCase(); - } + status = normalizeStatus(status); log.debug("设备状态: {} - {}", deviceName, status); - if ("offline".equals(status) && deviceDataHandler != null && deviceName != null) { - deviceDataHandler.handleDeviceOffline(deviceName); - } else if ("online".equals(status) && deviceDataHandler != null && deviceName != null) { - deviceDataHandler.handleDeviceOnline(deviceName); - } + handleDeviceStatusChange(json, deviceName, status); } /** @@ -241,17 +242,52 @@ public class AmqpMessageHandlerImpl implements AmqpMessageHandler { if (action == null) { action = json.getStr("status"); } - if (action != null) { - action = action.trim().toLowerCase(); - } + action = normalizeStatus(action); log.debug("设备{}: {}", "online".equals(action) ? "上线" : "下线", deviceName); - if ("offline".equals(action) && deviceDataHandler != null && deviceName != null) { + handleDeviceStatusChange(json, deviceName, action); + } + + private void handleDeviceStatusChange(JSONObject json, String deviceName, String status) { + if (deviceName == null || status == null) { + log.warn("设备状态消息缺少设备名称或状态: {}", json); + return; + } + + String productKey = json.getStr("productKey"); + if (deviceStatusService != null && productKey != null && !productKey.trim().isEmpty()) { + try { + Map statusData = new HashMap<>(); + statusData.put("productKey", productKey); + statusData.put("deviceName", deviceName); + statusData.put("status", status); + statusData.put("statusTime", json.get("statusTime")); + statusData.put("lastTime", json.get("lastTime")); + statusData.put("clientIp", json.getStr("clientIp")); + statusData.put("iotId", json.getStr("iotId")); + statusData.put("userId", json.get("userId")); + deviceStatusService.handleStatusChange(statusData); + return; + } catch (Exception e) { + log.error("写入设备状态失败，降级执行上下线处理: deviceName={}, status={}, err={}", + deviceName, status, e.getMessage(), e); + } + } + + if ("offline".equals(status) && deviceDataHandler != null) { deviceDataHandler.handleDeviceOffline(deviceName); - } else if ("online".equals(action) && deviceDataHandler != null && deviceName != null) { + } else if ("online".equals(status) && deviceDataHandler != null) { deviceDataHandler.handleDeviceOnline(deviceName); } } + private String normalizeStatus(String status) { + if (status == null) { + return null; + } + status = status.trim().toLowerCase(); + return ("online".equals(status) || "offline".equals(status)) ? status : null; + } + } diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceStatusServiceImpl.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceStatusServiceImpl.java index 3e92e19..3d5c2f5 100644 --- a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceStatusServiceImpl.java +++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceStatusServiceImpl.java @@ -2,6 +2,8 @@ package com.intc.iot.service.impl; import cn.hutool.json.JSONUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import com.intc.common.tenant.helper.TenantHelper; import com.intc.iot.handler.DeviceDataHandler; import com.intc.iot.domain.IotDeviceStatus; import com.intc.iot.mapper.IotDeviceStatusMapper; @@ -10,10 +12,11 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.dao.DuplicateKeyException; import org.springframework.stereotype.Service; -import org.springframework.transaction.annotation.Transactional; import java.time.LocalDateTime; +import java.util.List; import java.util.Map; /** @@ -34,30 +37,36 @@ public class DeviceStatusServiceImpl implements DeviceStatusService { private DeviceDataHandler deviceDataHandler; @Override - @Transactional(rollbackFor = Exception.class) public void handleStatusChange(Map statusData) { try { - String productKey = (String) statusData.get("productKey"); - String deviceName = (String) statusData.get("deviceName"); - String status = (String) statusData.get("status"); + if (statusData == null || statusData.isEmpty()) { + log.warn("[设备状态] 状态数据为空，跳过处理"); + return; + } + String productKey = trimToNull(statusData.get("productKey")); + String deviceName = trimToNull(statusData.get("deviceName")); + String status = normalizeStatus(statusData.get("status")); Long statusTime = getLongValue(statusData, "statusTime"); Long lastTime = getLongValue(statusData, "lastTime"); - String clientIp = (String) statusData.get("clientIp"); + String clientIp = trimToNull(statusData.get("clientIp")); + String iotId = trimToNull(statusData.get("iotId")); + Long userId = getLongValue(statusData, "userId"); if (productKey == null || deviceName == null || status == null) { log.warn("[设备状态] 状态数据不完整，跳过处理: {}", JSONUtil.toJsonStr(statusData)); return; } + if (statusTime == null) { + statusTime = System.currentTimeMillis() / 1000; + } + if (lastTime == null) { + lastTime = statusTime; + } log.info("[设备状态] 设备状态变更 - ProductKey: {}, DeviceName: {}, Status: {}", productKey, deviceName, status); - // 查询是否已存在记录 - IotDeviceStatus existingStatus = deviceStatusMapper.selectOne( - new LambdaQueryWrapper() - .eq(IotDeviceStatus::getProductKey, productKey) - .eq(IotDeviceStatus::getDeviceName, deviceName) - ); + IotDeviceStatus existingStatus = findExistingStatus(productKey, deviceName); LocalDateTime now = LocalDateTime.now(); @@ -67,8 +76,12 @@ public class DeviceStatusServiceImpl implements DeviceStatusService { existingStatus.setStatusTime(statusTime); existingStatus.setLastTime(lastTime); existingStatus.setClientIp(clientIp); + existingStatus.setIotId(iotId); + existingStatus.setUserId(userId); existingStatus.setUpdatedTime(now); - deviceStatusMapper.updateById(existingStatus); + TenantHelper.ignore(() -> { + deviceStatusMapper.updateById(existingStatus); + }); log.debug("[设备状态] 更新设备状态记录 - ID: {}, Status: {}", existingStatus.getId(), status); } else { @@ -76,21 +89,33 @@ public class DeviceStatusServiceImpl implements DeviceStatusService { IotDeviceStatus newStatus = new IotDeviceStatus(); newStatus.setProductKey(productKey); newStatus.setDeviceName(deviceName); - newStatus.setIotId((String) statusData.get("iotId")); + newStatus.setIotId(iotId); newStatus.setStatus(status); newStatus.setStatusTime(statusTime); newStatus.setLastTime(lastTime); newStatus.setClientIp(clientIp); + newStatus.setUserId(userId); newStatus.setCreatedTime(now); newStatus.setUpdatedTime(now); - deviceStatusMapper.insert(newStatus); + try { + TenantHelper.ignore(() -> { + deviceStatusMapper.insert(newStatus); + }); + } catch (DuplicateKeyException e) { + log.warn("[设备状态] 状态记录并发插入冲突，改为更新: productKey={}, deviceName={}", productKey, deviceName); + updateStatusByDevice(productKey, deviceName, newStatus); + } log.info("[设备状态] 新增设备状态记录 - ProductKey: {}, DeviceName: {}, Status: {}", productKey, deviceName, status); } - if ("online".equalsIgnoreCase(status) && deviceDataHandler != null) { - deviceDataHandler.handleDeviceOnline(deviceName); + if (deviceDataHandler != null) { + if ("online".equals(status)) { + deviceDataHandler.handleDeviceOnline(deviceName); + } else if ("offline".equals(status)) { + deviceDataHandler.handleDeviceOffline(deviceName); + } } } catch (Exception e) { log.error("[设备状态] 处理设备状态变更异常: {}", e.getMessage(), e); @@ -100,11 +125,11 @@ public class DeviceStatusServiceImpl implements DeviceStatusService { @Override public IotDeviceStatus queryDeviceStatus(String productKey, String deviceName) { - return deviceStatusMapper.selectOne( + return TenantHelper.ignore(() -> deviceStatusMapper.selectOne( new LambdaQueryWrapper() .eq(IotDeviceStatus::getProductKey, productKey) .eq(IotDeviceStatus::getDeviceName, deviceName) - ); + )); } @Override @@ -130,4 +155,53 @@ public class DeviceStatusServiceImpl implements DeviceStatusService { return null; } } + + private IotDeviceStatus findExistingStatus(String productKey, String deviceName) { + return TenantHelper.ignore(() -> { + List list = deviceStatusMapper.selectList( + new LambdaQueryWrapper() + .eq(IotDeviceStatus::getProductKey, productKey) + .eq(IotDeviceStatus::getDeviceName, deviceName) + ); + if (list == null || list.isEmpty()) { + return null; + } + if (list.size() > 1) { + log.warn("[设备状态] 存在重复状态记录: productKey={}, deviceName={}, count={}", + productKey, deviceName, list.size()); + } + return list.get(0); + }); + } + + private void updateStatusByDevice(String productKey, String deviceName, IotDeviceStatus status) { + TenantHelper.ignore(() -> { + deviceStatusMapper.update(status, + new LambdaUpdateWrapper() + .eq(IotDeviceStatus::getProductKey, productKey) + .eq(IotDeviceStatus::getDeviceName, deviceName) + ); + }); + } + + private String normalizeStatus(Object status) { + String text = trimToNull(status); + if (text == null) { + return null; + } + text = text.toLowerCase(); + if ("online".equals(text) || "offline".equals(text)) { + return text; + } + log.warn("[设备状态] 未知状态值，跳过处理: {}", text); + return null; + } + + private String trimToNull(Object value) { + if (value == null) { + return null; + } + String text = value.toString().trim(); + return text.isEmpty() ? null : text; + } }