qdintc/fishery-back
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 太阳能控制下发指令，自测bug修复。

Browse Source
This commit is contained in:
tianyongbao
2026-07-01 09:59:51 +08:00
parent c37d5ae5e3
commit 0c834efe0e
6 changed files with 220 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
intc-modules/intc-fishery/src/main/java/com/intc/fishery/domain/vo/DeviceVo.java
View File

@@ -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-已过期）
*/

46
intc-modules/intc-fishery/src/main/java/com/intc/fishery/service/impl/DeviceServiceImpl.java
View File

@@ -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<Device> buildQueryWrapper(DeviceBo bo) {
Map<String, Object> params = bo.getParams();
MPJLambdaWrapper<Device> 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<Device> wrapper, Map<String, Object> 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"));
}
}
/**
* 处理额外的查询参数
*

25
intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java
View File

@@ -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());
}
}
}
}

62
intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/AmqpMessageHandlerImpl.java
View File

@@ -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<String, Object> 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;
}
}

112
intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceStatusServiceImpl.java
View File

@@ -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<String, Object> 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<IotDeviceStatus>()
.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<IotDeviceStatus>()
.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<IotDeviceStatus> list = deviceStatusMapper.selectList(
new LambdaQueryWrapper<IotDeviceStatus>()
.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<IotDeviceStatus>()
.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;
}
}