Compare commits

58 Commits

Author SHA1 Message Date
tianyongbao
143dcd8e63 fix: 绑定功能优化完善,新增清零功能接口。 2026-07-11 19:16:33 +08:00
tianyongbao
818b59046d fix: 控制器定时任务,同一个控制器的开关同一时间下发的话,间隔20秒下发一条指令。 2026-07-06 21:42:07 +08:00
tianyongbao
2bcb2824b5 fix: 开关改变塘口时,不对定时任务清除,保留原来状态。 2026-07-04 22:13:25 +08:00
tianyongbao
0f320472cf fix: 下发开关定时任务,去掉开关状态判断,直接下发。 2026-07-02 07:35:00 +08:00
tianyongbao
81c6e7c7cb fix:太阳能网控,属性下发,不区分在线离线。 2026-07-01 23:53:23 +08:00
tianyongbao
ff6b2fe903 fix:太阳能网控,校准,走缓存逻辑。 2026-07-01 23:17:01 +08:00
tianyongbao
b1ed8a4152 fix:太阳能网控下发盐度设置,增加30分钟冻结。 2026-07-01 23:03:06 +08:00
tianyongbao
162fb7ed03 fix: 修改太阳能网控延时5s后,下发指令。 2026-07-01 16:59:23 +08:00
tianyongbao
71cbf04434 fix: 开关定时任务,扫描间隔从5分钟改为2分钟。 2026-07-01 13:15:49 +08:00
tianyongbao
9f57e796ec fix: 开关定时任务,bug修复。 2026-07-01 11:35:26 +08:00
tianyongbao
97b7cb9119 fix: 太阳能控制下发指令,自测bug修复。 2026-07-01 10:00:47 +08:00
tianyongbao
2930829ee9 feat: 新增太阳能控制下发指令功能。 2026-07-01 01:02:29 +08:00
tianyongbao
573bac7505 fix: 开关定时任务,后台定时扫描执行。 2026-06-30 17:26:51 +08:00
tianyongbao
f0f6494574 fix: 盐度设置,设备上线后,直接下发,去掉原来的1.5秒延时。 2026-06-30 10:32:31 +08:00
tianyongbao
257f19c16f fix: 盐度设置,设备上线后,延时1.5秒后再下发。 2026-06-29 23:36:09 +08:00
tianyongbao
5e55be6526 fix: 盐度设置,Redis失效问题解决。 2026-06-29 22:54:34 +08:00
tianyongbao
c7a1775aec fix: 盐度设置,属性上报时,也进行盐度设置。 2026-06-29 22:00:34 +08:00
tianyongbao
af022e63df fix: 盐度设置,功能优化,打印日志。 2026-06-29 10:40:00 +08:00
tianyongbao
4847f6a9e9 fix: 盐度补偿设定,太阳能网控缓存逻辑修改完善。 2026-06-28 22:52:33 +08:00
tianyongbao
3b37f59aff fix: 盐度补偿设定值同步到设备表中。 2026-06-24 19:24:33 +08:00
tianyongbao
ca1c0bc3eb fix: 欠压和欠流,不再拨打电话,功能修改。 2026-06-18 13:46:47 +08:00
tianyongbao
c9ac005953 fix: 定时控制,单个开关,不能超过5个。 2026-06-17 20:21:56 +08:00
tianyongbao
646574a02f fix: 用户修改手机号的时候,对应的用户的userName也需要同步修改。 2026-06-15 12:09:25 +08:00
tianyongbao
aa8ba2c0cf fix: 控制器新增报警开关控制,过期提醒只在8:00~21:00打电话。 2026-06-15 09:13:14 +08:00
tianyongbao
045f94f8d7 fix: 太阳能网控新增时候,离线时置离线有问题,bug修复。 2026-06-10 23:30:13 +08:00
tianyongbao
f4d2b4688b fix: 控制器三相电源380V时,默认开关输出电压改为220。 2026-06-06 18:36:47 +08:00
tianyongbao
bf2b2045c8 fix: 控制器没有塘口时,也能打电话产生报警。 2026-06-06 09:22:10 +08:00
tianyongbao
266fd0587a fix: 设备最后上数时间,历史数据同步接口修改。 2026-06-04 14:57:35 +08:00
tianyongbao
0ae4dd2034 fix: 设备最后上数时间,新增字段,接口调用,小程序显示。 2026-06-04 13:53:44 +08:00
tianyongbao
d70076ac16 fix: 微信支付成功,未更新设备有效期问题修复。 2026-06-03 23:54:46 +08:00
tianyongbao
bc36250ccf fix: 用户报警第一电话,逻辑判断修改,可任意修改,不需要必须包含自己手机号。 2026-06-03 22:02:13 +08:00
tianyongbao
fabfca7c94 fix: 微信支付问题,新增补偿机制。 2026-06-01 00:17:52 +08:00
tianyongbao
ceb3886729 fix: 设备即将过期,告警类型改为1,新增。 2026-05-30 19:44:58 +08:00
tianyongbao
ff74a6599e fix: 设备即将过期,功能优化完善。 2026-05-30 19:33:20 +08:00
tianyongbao
ec0237e448 fix: 设备24小时内到期,电话通知修改。 2026-05-29 22:10:06 +08:00
tianyongbao
0e306596e3 fix:设备过期,离线报警和溶解氧报警应不报警,sensorErrorCode逻辑问题修复。 2026-05-29 12:11:02 +08:00
tianyongbao
fbfacc065b fix: 设备添加,默认告警码为0,告警消息排序修改,sensorErrorCode处理修改。 2026-05-12 06:56:27 +08:00
tianyongbao
28891f03ff fix: 设备解绑后,新生成一个无用户无塘口的设备信息。 2026-05-12 05:49:38 +08:00
tianyongbao
b1aeb5b058 fix: 新绑定设备,偶尔出现设备名称重复情况,问题修复。 2026-05-06 23:04:12 +08:00
tianyongbao
09c735ab91 fix: 告警电话记录和报警记录,新增告警标题查询条件。 2026-05-06 22:50:13 +08:00
tianyongbao
7edf220e0a fix: 设备绑定记录,设备管理,联动控制等,设备编号支持模糊查询。 2026-05-02 22:49:36 +08:00
tianyongbao
feedc121db fix: 溶解氧探头离线告警修复。 2026-05-02 22:19:04 +08:00
tianyongbao
2adba6c12f fix: 太阳能网控,设备休眠状态修改。 2026-04-27 16:48:53 +08:00
tianyongbao
78daff5f34 fix: 太阳能网控,离线报警问题修复。 2026-04-26 23:01:35 +08:00
tianyongbao
a12cc4c905 fix: 测试服务,关闭vms回执。 2026-04-26 21:20:02 +08:00
tianyongbao
b36a92c981 fix: 电话报警,回执超时问题修复。 2026-04-26 21:19:25 +08:00
tianyongbao
303cd01a59 fix: 控制器定时控制删除时,实际未删除。 2026-04-26 11:19:10 +08:00
tianyongbao
467c05265f fix:太阳能网控,离线报警优化配置。 2026-04-26 07:25:03 +08:00
tianyongbao
65bcbf470c fix:太阳能网控,溶解氧下限未保存。 2026-04-26 06:27:00 +08:00
tianyongbao
7f1d1381ef fix:太阳能网控,后台代码修改完善。 2026-04-25 06:50:11 +08:00
tianyongbao
2bd71701a8 fix:子账号不充值后,回调后,更新设备有效期,问题修复。 2026-04-21 13:11:43 +08:00
tianyongbao
6b7b330934 fix:子账号不存在,提示信息修改。 2026-04-20 12:56:16 +08:00
tianyongbao
5c13c88c41 fix:设备电话通知和告警,以及设备历史数据,增加设备有效期判断。 2026-04-20 09:44:52 +08:00
tianyongbao
24a1e3e348 fix:给设备充值,去掉设备归属用户限制,允许给子账号充值。 2026-04-18 16:07:13 +08:00
tianyongbao
521c1ca344 fix:校准记录,去掉权限校验,登录用户都可以调用。 2026-04-15 16:33:37 +08:00
tianyongbao
b58f9c3b42 fix:告警类型错误,问题修复。 2026-04-14 22:45:05 +08:00
tianyongbao
07495f9270 fix:三相输入缺相,缺相无告警,bug修复。 2026-04-14 22:31:17 +08:00
tianyongbao
9eb2ec90c6 fix: 电话通知逻辑,功能修改,完善机制。 2026-04-14 22:31:14 +08:00
50 changed files with 4181 additions and 583 deletions

View File

@@ -229,7 +229,8 @@ aliyun:
# VMS 语音通知配置
vms:
# 是否启用 MNS 回扇消费true 时自动启动 MNS 消费线程接收 VMS 回扇)
mns-enabled: true
# 注意:测试服设为 false避免与正式服共抢同一个 MNS 队列导致回执随机丢失
mns-enabled: false
# VMS 语音服务对应的 AccessKey ID如果与上面 IoT 平台的 AK 相同可不填)
access-key-id: LTAI5tKh6fEKJJESaTdnChkh
# VMS 语音服务对应的 AccessKey Secret
@@ -245,6 +246,8 @@ aliyun:
call-success-suppress-hours: 2
# 设备离线告警延迟触发时间(分钟),设备离线超过此时间后才发送告警
offline-warn-delay-minutes: 6
# 太阳能网控离线告警延迟触发时间分钟工作1分钟休眠同分钟为正常行为需要更长延迟
solar-offline-warn-delay-minutes: 30
# 是否启用报警电话通知(全局开关,关闭后所有类型的报警都不会触发电话通知)
call-notice-enabled: false
# AMQP 服务端订阅配置(使用数据同步的 AppKey + AppSecret

View File

@@ -138,6 +138,7 @@ tenant:
- sys_user_role
- sys_client
- sys_oss_config
- iot_device_status
- flow_spel
# MyBatisPlus配置
@@ -221,6 +222,7 @@ xss:
# 排除链接
excludeUrls:
- /system/notice
- /weixin/pay_notify
--- # 分布式锁 lock4j 全局配置
lock4j:

View File

@@ -107,6 +107,14 @@
<artifactId>intc-system</artifactId>
</dependency>
<!-- TDengine 模块(用于清理设备时序数据) -->
<dependency>
<groupId>com.intc</groupId>
<artifactId>intc-tdengine</artifactId>
<version>5.5.0</version>
<scope>compile</scope>
</dependency>
</dependencies>
</project>

View File

@@ -66,6 +66,13 @@ public class DefineDeviceWarnCode {
*/
public static final int SaturaHigh = 1 << 10;
/**
* 设备休眠(太阳能网控专用) 1 << 11 = 2048
* 太阳能网控设备正常工作周期为 在线2分钟+休眠8分钟
* 休眠期间 IoT 平台上报离线,但属于正常行为,不等同于真正离线。
*/
public static final int DeviceSleeping = 1 << 11;
/**
* 设备断电(控制器) 1 << 15 = 32768
*/
@@ -91,7 +98,8 @@ public class DefineDeviceWarnCode {
/**
* 转换告警码为告警描述(与 C# ToWarnDescription 保持一致)
* 优先级:设备过期 > 设备离线 > 设备断电 > 溶解氧离线 > 溶解氧告警 > 温度告警 > 饱和度过高
* 优先级:设备过期 > 设备离线 > 设备休眠 > 设备断电 > 溶解氧离线 > 溶解氧告警 > 温度告警 > 饱和度过高
* 注PH/盐度/浊度/水位/氨氮探头离线不需要前端展示,故不在此处理
*/
public static String toWarnDescription(int warnCode) {
if (warnCode == None) {
@@ -103,6 +111,9 @@ public class DefineDeviceWarnCode {
if ((warnCode & DeviceOffline) != 0) {
return "设备离线";
}
if ((warnCode & DeviceSleeping) != 0) {
return "设备休眠";
}
if ((warnCode & ErrorPowerOff) != 0) {
return "设备断电";
}

View File

@@ -136,18 +136,18 @@ public class AquUserController extends BaseController {
if (userId == null || userId < 0) {
return R.fail("用户未登录");
}
// 查询用户的报警电话 JSON字段
AquUserVo user = aquUserService.queryById(userId);
if (user == null) {
return R.fail("用户不存在");
}
String warnPhoneJson = user.getWarnPhoneJson();
if (StringUtils.isEmpty(warnPhoneJson)) {
return R.fail("不存在报警电话列表");
}
// 反序列化JSON为List<String>
try {
List<String> warnPhones = JsonUtils.parseArray(warnPhoneJson, String.class);
@@ -156,7 +156,7 @@ public class AquUserController extends BaseController {
return R.fail("报警电话数据格式错误");
}
}
/**
* 获取子用户列表
*
@@ -169,10 +169,10 @@ public class AquUserController extends BaseController {
if (userId == null || userId < 0) {
return R.fail("用户未登录");
}
// 查询该用户的所有子账号
List<UserChildVo> list = userRelationMapper.selectChildUsersByParentId(userId);
return R.ok(list);
}
@@ -188,10 +188,10 @@ public class AquUserController extends BaseController {
if (userId == null || userId < 0) {
return R.fail("用户未登录");
}
// 查询该用户的所有父账号
List<UserChildVo> list = userRelationMapper.selectParentUsersByChildId(userId);
return R.ok(list);
}
@@ -215,11 +215,11 @@ public class AquUserController extends BaseController {
.eq(AquUser::getMobilePhone, request.getMobilePhone())
.select(AquUser::getId)
);
if (childUser == null) {
return R.fail("该手机号用户不存在");
return R.fail("该手机号还未注册小程序,请先注册");
}
Long childUserId = childUser.getId();
// 检查是否已存在该子账号关系
@@ -228,7 +228,7 @@ public class AquUserController extends BaseController {
.eq(UserRelation::getParentUserId, userId)
.eq(UserRelation::getChildUserId, childUserId)
);
if (count > 0) {
return R.fail("子账号已添加,无需重复添加");
}
@@ -295,30 +295,13 @@ public class AquUserController extends BaseController {
return R.fail("用户未登录");
}
// 查询用户信息,获取当前用户的手机号
// 查询用户信息,确认当前用户存在
AquUserVo user = aquUserService.queryById(userId);
if (user == null) {
return R.fail("用户不存在");
}
String oldMobilePhone = user.getMobilePhone();
if (StringUtils.isEmpty(oldMobilePhone)) {
return R.fail("用户手机号不存在,无权操作");
}
// 验证报警电话列表是否包含用户自己的手机号
List<String> listPhone = request.getListPhone();
boolean containUser = false;
for (String phone : listPhone) {
if (phone.equals(oldMobilePhone)) {
containUser = true;
break;
}
}
if (!containUser) {
return R.fail("报警电话必须包含自己手机号");
}
// 序列化为JSON字符串
String warnPhoneJson = JsonUtils.toJsonString(listPhone);
@@ -455,6 +438,7 @@ public class AquUserController extends BaseController {
// 同步更新 SysUser 中的 phonenumber
SysUserBo sysUserBo = new SysUserBo();
sysUserBo.setUserId(userId);
sysUserBo.setUserName(request.getNewMobilePhone());
sysUserBo.setPhonenumber(request.getNewMobilePhone());
sysUserService.updateUserProfile(sysUserBo);

View File

@@ -1,6 +1,7 @@
package com.intc.fishery.controller;
import java.util.List;
import java.util.ArrayList;
import com.intc.fishery.domain.bo.*;
import lombok.RequiredArgsConstructor;
@@ -29,18 +30,25 @@ import com.intc.fishery.domain.vo.PublicDeviceLinkedCtrl;
import com.intc.fishery.domain.vo.PublicDeviceDeadInfo;
import com.intc.fishery.domain.vo.PublicDeviceTypeAndSerialNum;
import com.intc.fishery.domain.Device;
import com.intc.fishery.domain.DeviceBindRecord;
import com.intc.fishery.domain.DeviceSwitch;
import com.intc.fishery.domain.DeviceErrorCode;
import com.intc.fishery.domain.Pond;
import com.intc.fishery.domain.LinkedCtrl;
import com.intc.fishery.domain.DeviceCorrectRecord;
import com.intc.fishery.mapper.DeviceMapper;
import com.intc.fishery.mapper.DeviceSwitchMapper;
import com.intc.fishery.mapper.DeviceBindRecordMapper;
import com.intc.fishery.mapper.DeviceErrorCodeMapper;
import com.intc.fishery.mapper.PondMapper;
import com.intc.fishery.mapper.LinkedCtrlMapper;
import com.intc.fishery.mapper.DeviceCorrectRecordMapper;
import com.intc.fishery.constant.DefineDeviceWarnCode;
import com.intc.fishery.utils.MessageOpRecordUtil;
import com.intc.common.core.config.properties.DeviceTypeProperties;
import com.intc.common.redis.utils.RedisUtils;
import com.intc.tdengine.service.IDeviceSensorDataService;
import org.springframework.beans.factory.annotation.Autowired;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import org.springframework.transaction.annotation.Transactional;
@@ -68,8 +76,13 @@ public class DeviceController extends BaseController {
private final PondMapper pondMapper;
private final LinkedCtrlMapper linkedCtrlMapper;
private final DeviceCorrectRecordMapper deviceCorrectRecordMapper;
private final DeviceBindRecordMapper deviceBindRecordMapper;
private final DeviceErrorCodeMapper deviceErrorCodeMapper;
private final DeviceTypeProperties deviceTypeProperties;
@Autowired(required = false)
private IDeviceSensorDataService deviceSensorDataService;
/**
* 查询设备管理列表
*/
@@ -79,6 +92,16 @@ public class DeviceController extends BaseController {
return deviceService.queryPageList(bo, pageQuery);
}
/**
* 查询太阳能网控设备列表
*/
@SaCheckPermission("fishery:solarControl:list")
@GetMapping("/solar/list")
public TableDataInfo<DeviceVo> solarList(DeviceBo bo, PageQuery pageQuery) {
bo.setDeviceType(3);
return deviceService.queryPageList(bo, pageQuery);
}
/**
* 导出设备管理列表
*/
@@ -137,6 +160,106 @@ public class DeviceController extends BaseController {
return toAjax(deviceService.deleteWithValidByIds(List.of(ids), true));
}
/**
* 清零设备测试数据(出厂测试用)
* <p>清理范围:
* <ul>
* <li>有效期(deadTime)、告警状态(warnCode)、实时传感器值、最后上数时间</li>
* <li>TDengine时序数据删除设备子表清除所有历史传感器数据</li>
* <li>设备绑定/解绑记录</li>
* <li>设备校准记录</li>
* <li>设备故障码记录</li>
* <li>Redis告警标记</li>
* </ul>
* 清零后设备重新绑定时会自动重新计算365天有效期。</p>
*
* @param id 设备ID
* @return 操作结果
*/
@SaCheckPermission("fishery:device:edit")
@Log(title = "设备管理", businessType = BusinessType.UPDATE)
@RepeatSubmit()
@PutMapping("/resetTestData/{id}")
public R<Void> resetTestData(@NotNull(message = "主键不能为空")
@PathVariable Long id) {
Device device = deviceMapper.selectById(id);
if (device == null) {
return R.fail("设备不存在");
}
String serialNum = device.getSerialNum();
// 1. 清零设备表:有效期、告警状态、实时传感器值、最后上数时间
deviceMapper.update(null,
new LambdaUpdateWrapper<Device>()
.eq(Device::getId, id)
.set(Device::getDeadTime, null)
.set(Device::getWarnCode, DefineDeviceWarnCode.None)
.set(Device::getLastReportTime, null)
.set(Device::getValueDissolvedOxygen, null)
.set(Device::getValueTemperature, null)
.set(Device::getValueSaturability, null)
.set(Device::getValuePh, null)
.set(Device::getValueSalinity, null)
.set(Device::getValueBattery, null)
.set(Device::getIsOxygenWarnExist, 0)
.set(Device::getIsTempWarnExist, 0)
);
// 2. 删除TDengine时序数据设备子表含所有历史传感器数据
if (deviceSensorDataService != null && serialNum != null && !serialNum.isBlank()) {
try {
deviceSensorDataService.dropSubTable(serialNum);
} catch (Exception e) {
// TDengine清理失败不阻断主流程
org.slf4j.LoggerFactory.getLogger(DeviceController.class)
.warn("清理TDengine数据失败: deviceId={}, serialNum={}, error={}", id, serialNum, e.getMessage());
}
}
// 3. 删除设备绑定/解绑记录
deviceBindRecordMapper.delete(
new LambdaQueryWrapper<DeviceBindRecord>()
.eq(DeviceBindRecord::getSerialNum, serialNum)
);
// 4. 删除设备校准记录
deviceCorrectRecordMapper.delete(
new LambdaQueryWrapper<DeviceCorrectRecord>()
.eq(DeviceCorrectRecord::getDeviceId, id)
);
// 5. 删除设备故障码记录
deviceErrorCodeMapper.delete(
new LambdaQueryWrapper<DeviceErrorCode>()
.eq(DeviceErrorCode::getDeviceId, id)
);
// 6. 清除Redis告警标记
if (serialNum != null && !serialNum.isBlank()) {
try {
RedisUtils.deleteObject("device:offline:wait:" + serialNum);
RedisUtils.deleteObject("device:detector:oxy:offline:wait:" + serialNum);
} catch (Exception e) {
// Redis清理失败不影响主流程
}
}
// 7. 记录操作日志
if (device.getUserId() != null) {
MessageOpRecordUtil.addMessageOpRecordUser(
device.getUserId(),
"设备管理",
String.format("%s(%s),清零设备测试数据。", device.getDeviceName(), serialNum)
);
} else {
org.slf4j.LoggerFactory.getLogger(DeviceController.class)
.info("清零设备测试数据(设备未绑定用户): deviceId={}, serialNum={}", id, serialNum);
}
return R.ok();
}
/**
* 查询用户所有设备简化列表
* type=1: 返回完整信息(包含塘口信息和开关列表)
@@ -317,6 +440,11 @@ public class DeviceController extends BaseController {
data.setSalinityCompensation(device.getSalinityCompensation());
data.setInputVoltage(device.getInputVoltage());
data.setVoltageWarnOpen(device.getVoltageWarnOpen());
data.setControllerFaultCallOpen(device.getControllerFaultCallOpen() == null ? 1 : device.getControllerFaultCallOpen());
data.setBatteryWarnCallOpen(device.getBatteryWarnCallOpen());
data.setBatteryWarnCallNoDis(device.getBatteryWarnCallNoDis());
data.setBatteryWarnLower(device.getBatteryWarnLower());
// 设置塘口信息
if (device.getPondId() != null && device.getPondId() > 0) {
@@ -820,46 +948,66 @@ public class DeviceController extends BaseController {
}
/**
* 查询即将到期设备列表
* 返回当前时间到未来1个月内将要到期的设备
* 查询设备过期列表
* 返回已过期和即将过期未最1个月内的设备
*
* @param rootUserId 用户ID
* @return 即将到期的设备列表
* @return 设备过期列表(包含已过期和即将过期)
*/
@GetMapping("/list_device_dead")
public R<List<PublicDeviceDeadInfo>> queryListDeviceDead(
@RequestParam("rootUserId") Long rootUserId) {
// 计算时间范围:当前时间到未1个月
// 计算时间范围:当前时间到未1个月
Date now = new Date();
Calendar calendar = Calendar.getInstance();
calendar.setTime(now);
calendar.add(Calendar.MONTH, 1);
Date oneMonthLater = calendar.getTime();
// 查询即将到期的设备
List<Device> devices = deviceMapper.selectList(
List<PublicDeviceDeadInfo> list = new ArrayList<>();
// 查询已过期的设备deadTime <= 当前时间)
List<Device> expiredDevices = deviceMapper.selectList(
new LambdaQueryWrapper<Device>()
.eq(Device::getUserId, rootUserId)
.le(Device::getDeadTime, now)
.select(Device::getId, Device::getDeviceName, Device::getSerialNum, Device::getDeadTime)
.orderByAsc(Device::getDeadTime)
);
// 查询即将过期的设备(当前时间 < deadTime <= 未最1个月
List<Device> soonExpireDevices = deviceMapper.selectList(
new LambdaQueryWrapper<Device>()
.eq(Device::getUserId, rootUserId)
.gt(Device::getDeadTime, now)
.le(Device::getDeadTime, oneMonthLater)
.ge(Device::getWarnCode, 0)
.select(Device::getId, Device::getDeviceName, Device::getSerialNum, Device::getDeadTime)
.orderByAsc(Device::getDeadTime)
);
// 转换为VO列表
List<PublicDeviceDeadInfo> list = devices.stream()
.map(device -> {
PublicDeviceDeadInfo vo = new PublicDeviceDeadInfo();
vo.setId(device.getId());
vo.setDeviceName(device.getDeviceName());
vo.setSerialNum(device.getSerialNum());
vo.setDeadTime(device.getDeadTime());
return vo;
})
.collect(Collectors.toList());
// 转换已过期设备status=1
for (Device device : expiredDevices) {
PublicDeviceDeadInfo vo = new PublicDeviceDeadInfo();
vo.setId(device.getId());
vo.setDeviceName(device.getDeviceName());
vo.setSerialNum(device.getSerialNum());
vo.setDeadTime(device.getDeadTime());
vo.setStatus(1);
list.add(vo);
}
// 转换即将过期设备status=2
for (Device device : soonExpireDevices) {
PublicDeviceDeadInfo vo = new PublicDeviceDeadInfo();
vo.setId(device.getId());
vo.setDeviceName(device.getDeviceName());
vo.setSerialNum(device.getSerialNum());
vo.setDeadTime(device.getDeadTime());
vo.setStatus(2);
list.add(vo);
}
return R.ok(list);
}
@@ -1015,6 +1163,62 @@ public class DeviceController extends BaseController {
return R.ok();
}
/**
* 设置控制器故障电话报警开关。
* 仅控制缺相、过压、欠压、开关空载/欠流/过流等故障的电话通知;断电电话通知不受此开关限制。
*
* @param rootUserId 用户ID
* @param request 请求对象包含设备ID和开关状态
* @return 操作结果
*/
@PutMapping("/controller_fault_call_open")
public R<Void> updateControllerFaultCallOpen(
@RequestParam("rootUserId") Long rootUserId,
@Validated @RequestBody ReqTurnOpen request) {
Device device = deviceMapper.selectOne(
new LambdaQueryWrapper<Device>()
.eq(Device::getId, request.getId())
.select(Device::getId, Device::getUserId, Device::getSerialNum,
Device::getDeviceName, Device::getDeviceType,
Device::getControllerFaultCallOpen, Device::getDeadTime)
);
if (device == null || device.getUserId() == null || !device.getUserId().equals(rootUserId)
|| device.getDeviceType() == null || device.getDeviceType() != 2) {
return R.fail("设备不存在或无权限操作");
}
if (device.getDeadTime() != null && device.getDeadTime().before(new Date())) {
return R.fail("设备服务已过期");
}
Integer currentOpen = device.getControllerFaultCallOpen() == null ? 1 : device.getControllerFaultCallOpen();
if (currentOpen.equals(request.getIsOpen())) {
return R.ok();
}
boolean updated = deviceMapper.update(null,
new LambdaUpdateWrapper<Device>()
.eq(Device::getId, request.getId())
.set(Device::getControllerFaultCallOpen, request.getIsOpen())
) > 0;
if (!updated) {
return R.fail("更新失败");
}
String op = request.getIsOpen() == 1 ? "开启" : "关闭";
MessageOpRecordUtil.addMessageOpRecordUser(
rootUserId,
"设备告警设置",
String.format("%s(%s) 设置控制器故障电话报警开关:%s",
device.getDeviceName(), device.getSerialNum(), op)
);
return R.ok();
}
/**
* 设置测控一体机溶解氧启用/禁用
* 启用时:同时启用溶解氧和温度告警及免打扰

View File

@@ -186,7 +186,7 @@ public class PondController extends BaseController {
/**
* 根据塘口ID查询设备列表
* 返回数据按设备类型分类:
* - listDetector: 水质检测仪列表deviceType=1+ 开启氧检测的测控一体机deviceType=2 && isOxygenUsed=1
* - listDetector: 水质检测仪列表deviceType=1+ 开启氧检测的测控一体机deviceType=2 && isOxygenUsed=1+ 太阳能网控deviceType=3
* - listController: 测控一体机列表deviceType=2包含开关列表
*
* @param pondId 塘口ID
@@ -294,10 +294,11 @@ public class PondController extends BaseController {
// 9. 处理探测器列表(严格从 pondDevices 取,与 C# pond.ListDevice 保持一致,不包含仅开关挂塘口的设备)
List<DeviceVo> detectorList = new ArrayList<>();
for (Device device : pondDevices) {
// 水质检测仪 或 开启氧检测的测控一体机
// 水质检测仪 或 开启氧检测的测控一体机 或 太阳能网控
if ((device.getDeviceType() != null && device.getDeviceType() == 1)
|| (device.getDeviceType() != null && device.getDeviceType() == 2
&& device.getIsOxygenUsed() != null && device.getIsOxygenUsed() == 1)) {
&& device.getIsOxygenUsed() != null && device.getIsOxygenUsed() == 1)
|| (device.getDeviceType() != null && device.getDeviceType() == 3)) {
DeviceVo deviceVo = MapstructUtils.convert(device, DeviceVo.class);
@@ -305,7 +306,8 @@ public class PondController extends BaseController {
PondMode1WarnCodeInfo detectorWarnCodeInfo = new PondMode1WarnCodeInfo();
if (device.getWarnCode() != null && device.getWarnCode() < DefineDeviceWarnCode.UserMaxCode) {
detectorWarnCodeInfo.setWarnCode(device.getWarnCode());
detectorWarnCodeInfo.setWarnDescription(DefineDeviceWarnCode.toWarnDescription(device.getWarnCode()));
String desc = DefineDeviceWarnCode.toWarnDescription(device.getWarnCode());
detectorWarnCodeInfo.setWarnDescription(desc != null ? desc : "");
} else {
detectorWarnCodeInfo.setWarnCode(DefineDeviceWarnCode.None);
detectorWarnCodeInfo.setWarnDescription("");
@@ -356,7 +358,8 @@ public class PondController extends BaseController {
PondMode1WarnCodeInfo warnCodeInfo = new PondMode1WarnCodeInfo();
if (device.getWarnCode() != null && device.getWarnCode() < DefineDeviceWarnCode.UserMaxCode) {
warnCodeInfo.setWarnCode(device.getWarnCode());
warnCodeInfo.setWarnDescription(DefineDeviceWarnCode.toWarnDescription(device.getWarnCode()));
String desc = DefineDeviceWarnCode.toWarnDescription(device.getWarnCode());
warnCodeInfo.setWarnDescription(desc != null ? desc : "");
} else {
warnCodeInfo.setWarnCode(DefineDeviceWarnCode.None);
warnCodeInfo.setWarnDescription("");
@@ -808,7 +811,7 @@ public class PondController extends BaseController {
warnCode |= device.getWarnCode();
}
// 判断设备是否在线(简化判断,实际应根据业务逻辑)
// 判断设备是否在线DeviceSleeping 不算离线,太阳能网控休眠属于正常工作)
if (device.getWarnCode() == null
|| ((device.getWarnCode() & DefineDeviceWarnCode.DeviceTimeDead) == 0
&& (device.getWarnCode() & DefineDeviceWarnCode.DeviceOffline) == 0)) {
@@ -922,7 +925,8 @@ public class PondController extends BaseController {
data.setDeviceCount(hashSetDeviceId.size());
data.setSwitchCount(pondSwitches.size());
data.getWarnCodeInfo().setWarnCode(warnCode);
data.getWarnCodeInfo().setWarnDescription(DefineDeviceWarnCode.toWarnDescription(warnCode));
String warnDesc = DefineDeviceWarnCode.toWarnDescription(warnCode);
data.getWarnCodeInfo().setWarnDescription(warnDesc != null ? warnDesc : "");
listData.add(data);
}

View File

@@ -234,14 +234,14 @@ public class TimingCtrlController extends BaseController {
return R.fail("开关未绑定塘口");
}
// 检查定时控制数量是否超限4
// 检查定时控制数量是否超限5
long count = timingCtrlMapper.selectCount(
new LambdaQueryWrapper<TimingCtrl>()
.eq(TimingCtrl::getSwitchId, request.getId())
);
if (count >= 4) {
return R.fail("定时控制数量不得超过4");
if (count >= 5) {
return R.fail("定时控制数量不得超过5");
}
// 创建定时控制记录

View File

@@ -208,6 +208,11 @@ public class Device extends TenantEntity {
*/
private Integer voltageWarnOpen;
/**
* 控制器故障电话报警开关(断电不受此开关限制)
*/
private Integer controllerFaultCallOpen;
/**
* 备注
*/
@@ -238,5 +243,15 @@ public class Device extends TenantEntity {
*/
private Long batteryWarnLower;
/**
* 电量值太阳能网控专用0-100
*/
private Integer valueBattery;
/**
* 最后上数时间
*/
private Date lastReportTime;
}

View File

@@ -240,6 +240,11 @@ public class DeviceBo extends BaseEntity {
// @NotNull(message = "电压告警开关不能为空", groups = { AddGroup.class, EditGroup.class })
private Integer voltageWarnOpen;
/**
* 控制器故障电话报警开关(断电不受此开关限制)
*/
private Integer controllerFaultCallOpen;
/**
* 备注
*/
@@ -270,5 +275,10 @@ public class DeviceBo extends BaseEntity {
*/
private Long batteryWarnLower;
/**
* 最后上数时间
*/
private Date lastReportTime;
}

View File

@@ -270,6 +270,12 @@ public class DeviceVo implements Serializable {
@ExcelProperty(value = "电压告警开关")
private Integer voltageWarnOpen;
/**
* 控制器故障电话报警开关(断电不受此开关限制)
*/
@ExcelProperty(value = "控制器故障电话报警开关")
private Integer controllerFaultCallOpen;
/**
* 备注
*/
@@ -306,6 +312,18 @@ public class DeviceVo implements Serializable {
@ExcelProperty(value = "电量电话告警下限")
private Long batteryWarnLower;
/**
* 电量值太阳能网控专用0-100
*/
@ExcelProperty(value = "电量值")
private Integer valueBattery;
/**
* 最后上数时间
*/
@ExcelProperty(value = "最后上数时间")
private Date lastReportTime;
/**
* 手机号
*/
@@ -318,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-已过期)
*/

View File

@@ -109,6 +109,26 @@ public class PublicDeviceBaseData implements Serializable {
*/
private Integer voltageWarnOpen;
/**
* 控制器故障电话报警开关(断电不受此开关限制)
*/
private Integer controllerFaultCallOpen;
/**
* 低电量电话告警开关(太阳能网控专用)
*/
private Long batteryWarnCallOpen;
/**
* 低电量告警免打扰(太阳能网控专用)
*/
private Long batteryWarnCallNoDis;
/**
* 低电量告警下限太阳能网控专用0-100
*/
private Long batteryWarnLower;
/**
* 开关列表(仅测控一体机)
*/

View File

@@ -45,4 +45,10 @@ public class PublicDeviceDeadInfo implements Serializable {
@Schema(description = "服务到期时间")
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
private Date deadTime;
/**
* 过期状态1-已过期2-即将过期
*/
@Schema(description = "过期状态1-已过期2-即将过期")
private Integer status;
}

View File

@@ -171,6 +171,12 @@ public class CallNoticeServiceImpl implements ICallNoticeService {
wrapper.apply("mw.warn_type = {0}", warnType);
}
// 处理告警标题模糊查询
String warnTitle = (String) params.get("warnTitle");
if (StringUtils.isNotBlank(warnTitle)) {
wrapper.like("mw.title", warnTitle);
}
// 处理告警ID精确查询
Object messageWarnIdObj = params.get("messageWarnId");
if (messageWarnIdObj != null) {

View File

@@ -95,7 +95,7 @@ public class DeviceBindRecordServiceImpl implements IDeviceBindRecordService {
// 基础查询条件
wrapper.like(StringUtils.isNotBlank(bo.getIotId()), DeviceBindRecord::getIotId, bo.getIotId());
wrapper.eq(bo.getDeviceType() != null, DeviceBindRecord::getDeviceType, bo.getDeviceType());
wrapper.eq(StringUtils.isNotBlank(bo.getSerialNum()), DeviceBindRecord::getSerialNum, bo.getSerialNum());
wrapper.like(StringUtils.isNotBlank(bo.getSerialNum()), DeviceBindRecord::getSerialNum, bo.getSerialNum());
wrapper.eq(bo.getUserId() != null, DeviceBindRecord::getUserId, bo.getUserId());
wrapper.eq(bo.getIsBind() != null, DeviceBindRecord::getIsBind, bo.getIsBind());

View File

@@ -94,7 +94,7 @@ public class DeviceCorrectRecordServiceImpl implements IDeviceCorrectRecordServi
// 查询条件
wrapper.eq(bo.getDeviceId() != null, DeviceCorrectRecord::getDeviceId, bo.getDeviceId())
.eq(bo.getUserId() != null, DeviceCorrectRecord::getUserId, bo.getUserId())
.eq(StringUtils.isNotBlank(bo.getSerialNum()), DeviceCorrectRecord::getSerialNum, bo.getSerialNum())
.like(StringUtils.isNotBlank(bo.getSerialNum()), DeviceCorrectRecord::getSerialNum, bo.getSerialNum())
.eq(bo.getDeviceType() != null, DeviceCorrectRecord::getDeviceType, bo.getDeviceType())
.eq(bo.getValueDissolvedOxygen() != null, DeviceCorrectRecord::getValueDissolvedOxygen, bo.getValueDissolvedOxygen())
.eq(bo.getValueTemperature() != null, DeviceCorrectRecord::getValueTemperature, bo.getValueTemperature())

View File

@@ -5,10 +5,12 @@ 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;
import com.intc.common.mybatis.core.page.TableDataInfo;
import com.intc.common.redis.utils.RedisUtils;
import com.intc.fishery.domain.AquUser;
import com.intc.fishery.domain.Device;
import com.intc.fishery.domain.DeviceBindRecord;
@@ -20,6 +22,8 @@ import com.intc.fishery.domain.Pond;
import com.intc.fishery.domain.TimingCtrl;
import com.intc.fishery.domain.bo.DeviceBo;
import com.intc.fishery.domain.vo.DeviceVo;
import com.intc.fishery.constant.DefineDeviceWarnCode;
import com.intc.fishery.mapper.AquUserMapper;
import com.intc.fishery.mapper.DeviceBindRecordMapper;
import com.intc.fishery.mapper.DeviceCorrectRecordMapper;
import com.intc.fishery.mapper.DeviceErrorCodeMapper;
@@ -28,6 +32,7 @@ import com.intc.fishery.mapper.DeviceSwitchMapper;
import com.intc.fishery.mapper.LinkedCtrlMapper;
import com.intc.fishery.mapper.TimingCtrlMapper;
import com.intc.fishery.service.IDeviceService;
import com.intc.fishery.utils.MessageOpRecordUtil;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -49,12 +54,29 @@ import lombok.extern.slf4j.Slf4j;
public class DeviceServiceImpl implements IDeviceService {
private final DeviceMapper baseMapper;
private final AquUserMapper aquUserMapper;
private final DeviceBindRecordMapper deviceBindRecordMapper;
private final DeviceCorrectRecordMapper deviceCorrectRecordMapper;
private final DeviceErrorCodeMapper deviceErrorCodeMapper;
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 + ")";
/**
* 查询设备管理
@@ -105,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);
@@ -117,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;
@@ -128,7 +157,7 @@ public class DeviceServiceImpl implements IDeviceService {
}
wrapper.eq(bo.getUserId() != null, Device::getUserId, bo.getUserId());
wrapper.eq(StringUtils.isNotBlank(bo.getSerialNum()), Device::getSerialNum, bo.getSerialNum());
wrapper.like(StringUtils.isNotBlank(bo.getSerialNum()), Device::getSerialNum, bo.getSerialNum());
wrapper.like(StringUtils.isNotBlank(bo.getDeviceName()), Device::getDeviceName, bo.getDeviceName());
wrapper.eq(bo.getDeviceType() != null, Device::getDeviceType, bo.getDeviceType());
wrapper.eq(bo.getBindTime() != null, Device::getBindTime, bo.getBindTime());
@@ -144,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"));
}
}
/**
* 处理额外的查询参数
*
@@ -224,8 +280,18 @@ public class DeviceServiceImpl implements IDeviceService {
}
/**
* 校验并批量删除设备管理信息
* 删除顺序:故障码 -> 校准记录 -> 定时控制 -> 开关(先解除联动控制引用) -> 联动控制 -> 设备
* 校验并批量删除设备管理信息(参考 C# DeleteDevice 逻辑)
* <p>
* 处理流程:
* 1. 查询完整设备信息与用户信息
* 2. 新增解绑记录
* 3. 删除联动控制
* 4. 删除关联数据(故障码、校准记录、定时控制、开关)
* 5. 删除设备
* 6. 非管理员用户创建新的无绑定设备保留IoT信息
* 7. 清除告警数据warnCode + Redis离线等待标记
* 8. 记录操作日志
* </p>
*
* @param ids 待删除的主键集合
* @param isValid 是否进行有效性校验
@@ -234,85 +300,23 @@ public class DeviceServiceImpl implements IDeviceService {
@Override
@Transactional(rollbackFor = Exception.class)
public Boolean deleteWithValidByIds(Collection<Long> ids, Boolean isValid) {
if(isValid){
if (isValid) {
//TODO 做一些业务上的校验,判断是否需要校验
}
// 1. 删除设备关联的故障码记录
deviceErrorCodeMapper.delete(
new LambdaQueryWrapper<DeviceErrorCode>()
.in(DeviceErrorCode::getDeviceId, ids)
);
// 2. 删除设备关联的校准记录
deviceCorrectRecordMapper.delete(
new LambdaQueryWrapper<DeviceCorrectRecord>()
.in(DeviceCorrectRecord::getDeviceId, ids)
);
// 3. 查询设备关联的开关
List<DeviceSwitch> switches = deviceSwitchMapper.selectList(
new LambdaQueryWrapper<DeviceSwitch>()
.in(DeviceSwitch::getDeviceId, ids)
);
if (!switches.isEmpty()) {
// 收集开关ID
List<Long> switchIds = switches.stream()
.map(DeviceSwitch::getId)
.collect(Collectors.toList());
// 4. 删除开关关联的定时控制
timingCtrlMapper.delete(
new LambdaQueryWrapper<TimingCtrl>()
.in(TimingCtrl::getSwitchId, switchIds)
);
// 收集开关关联的联动控制ID
List<Long> linkedCtrlIds = switches.stream()
.map(DeviceSwitch::getLinkedCtrlId)
.filter(id -> id != null && id > 0)
.distinct()
.collect(Collectors.toList());
// 5. 解除开关与联动控制的绑定
deviceSwitchMapper.update(null,
new LambdaUpdateWrapper<DeviceSwitch>()
.set(DeviceSwitch::getLinkedCtrlId, null)
.in(DeviceSwitch::getDeviceId, ids)
);
// 6. 删除联动控制(如果有)
if (!linkedCtrlIds.isEmpty()) {
linkedCtrlMapper.delete(
new LambdaQueryWrapper<LinkedCtrl>()
.in(LinkedCtrl::getId, linkedCtrlIds)
);
}
// 7. 删除开关
deviceSwitchMapper.delete(
new LambdaQueryWrapper<DeviceSwitch>()
.in(DeviceSwitch::getDeviceId, ids)
);
}
// 8. 删除设备本身关联的联动控制
linkedCtrlMapper.delete(
new LambdaQueryWrapper<LinkedCtrl>()
.in(LinkedCtrl::getDeviceId, ids)
);
// 9. 查询要删除的设备信息,添加解绑记录
// 1. 先查询完整设备信息(删除前需要用到各字段)
List<Device> devicesToDelete = baseMapper.selectList(
new LambdaQueryWrapper<Device>()
.in(Device::getId, ids)
.select(Device::getId, Device::getUserId, Device::getIotId,
Device::getDeviceType, Device::getSerialNum)
);
if (devicesToDelete.isEmpty()) {
return false;
}
// 为每个设备添加解绑记录
for (Device device : devicesToDelete) {
Long deviceId = device.getId();
// 2. 新增解绑记录
try {
DeviceBindRecord unbindRecord = new DeviceBindRecord();
unbindRecord.setUserId(device.getUserId());
@@ -321,16 +325,141 @@ public class DeviceServiceImpl implements IDeviceService {
unbindRecord.setDeviceType(device.getDeviceType() != null ? Long.valueOf(device.getDeviceType()) : null);
unbindRecord.setSerialNum(device.getSerialNum());
deviceBindRecordMapper.insert(unbindRecord);
log.info("添加设备解绑记录: deviceId={}, userId={}, serialNum={}",
device.getId(), device.getUserId(), device.getSerialNum());
log.info("添加设备解绑记录: deviceId={}, userId={}, serialNum={}",
deviceId, device.getUserId(), device.getSerialNum());
} catch (Exception e) {
log.error("添加设备解绑记录失败: deviceId={}, error={}", device.getId(), e.getMessage(), e);
// 不影响删除流程,继续执行
log.error("添加设备解绑记录失败: deviceId={}, error={}", deviceId, e.getMessage(), e);
}
// 3. 删除设备关联的联动控制
linkedCtrlMapper.delete(
new LambdaQueryWrapper<LinkedCtrl>()
.eq(LinkedCtrl::getDeviceId, deviceId)
);
// 4. 删除设备关联的故障码记录
deviceErrorCodeMapper.delete(
new LambdaQueryWrapper<DeviceErrorCode>()
.eq(DeviceErrorCode::getDeviceId, deviceId)
);
// 5. 删除设备关联的校准记录
deviceCorrectRecordMapper.delete(
new LambdaQueryWrapper<DeviceCorrectRecord>()
.eq(DeviceCorrectRecord::getDeviceId, deviceId)
);
// 6. 查询设备关联的开关,删除定时控制、解除联动绑定后删除开关
List<DeviceSwitch> switches = deviceSwitchMapper.selectList(
new LambdaQueryWrapper<DeviceSwitch>()
.eq(DeviceSwitch::getDeviceId, deviceId)
);
if (!switches.isEmpty()) {
List<Long> switchIds = switches.stream()
.map(DeviceSwitch::getId)
.collect(Collectors.toList());
// 删除开关关联的定时控制
timingCtrlMapper.delete(
new LambdaQueryWrapper<TimingCtrl>()
.in(TimingCtrl::getSwitchId, switchIds)
);
// 收集开关关联的联动控制ID
List<Long> linkedCtrlIds = switches.stream()
.map(DeviceSwitch::getLinkedCtrlId)
.filter(id -> id != null && id > 0)
.distinct()
.collect(Collectors.toList());
// 解除开关与联动控制的绑定
deviceSwitchMapper.update(null,
new LambdaUpdateWrapper<DeviceSwitch>()
.set(DeviceSwitch::getLinkedCtrlId, null)
.eq(DeviceSwitch::getDeviceId, deviceId)
);
// 删除联动控制(如果有)
if (!linkedCtrlIds.isEmpty()) {
linkedCtrlMapper.delete(
new LambdaQueryWrapper<LinkedCtrl>()
.in(LinkedCtrl::getId, linkedCtrlIds)
);
}
// 删除开关
deviceSwitchMapper.delete(
new LambdaQueryWrapper<DeviceSwitch>()
.eq(DeviceSwitch::getDeviceId, deviceId)
);
}
// 7. 删除设备记录
baseMapper.deleteById(deviceId);
// 8. 用户创建新的无绑定设备保留IoT信息清除用户和塘口关联
if (device.getUserId() != null) {
try {
AquUser user = aquUserMapper.selectById(device.getUserId());
if (user != null) {
// 管理员(出厂测试人员)解绑时清空有效期,普通用户解绑时保留有效期(防止解绑续命)
boolean isManager = user.getIsManager() != null && user.getIsManager() == 1L;
Device newDevice = new Device();
newDevice.setUserId(null);
newDevice.setIsOxygenUsed(device.getIsOxygenUsed());
newDevice.setIotId(device.getIotId());
newDevice.setSerialNum(device.getSerialNum());
newDevice.setDeviceName(device.getDeviceName());
newDevice.setDeviceType(device.getDeviceType());
newDevice.setIccId(device.getIccId());
newDevice.setBindTime(device.getBindTime());
// 管理员解绑时清空有效期(出厂测试),普通用户解绑时保留有效期
newDevice.setDeadTime(isManager ? null : device.getDeadTime());
newDevice.setPondId(null);
newDevice.setSalinityCompensation(0.0);
newDevice.setOxyWarnLower(device.getOxyWarnLower());
newDevice.setOxyWarnCallOpen(0);
newDevice.setOxyWarnCallNoDis(0);
newDevice.setTempWarnCallOpen(0);
newDevice.setTempWarnCallNoDis(0);
newDevice.setControllerFaultCallOpen(1);
newDevice.setWarnCode(DefineDeviceWarnCode.None);
newDevice.setCategory(device.getCategory());
baseMapper.insert(newDevice);
log.info("已为用户重建无绑定设备: oldDeviceId={}, newDeviceId={}, serialNum={}, isManager={}, deadTime={}",
deviceId, newDevice.getId(), device.getSerialNum(), isManager, newDevice.getDeadTime());
}
} catch (Exception e) {
log.error("重建无绑定设备失败: deviceId={}, error={}", deviceId, e.getMessage(), e);
}
}
// 9. 清除告警数据Redis离线等待标记
try {
String serialNum = device.getSerialNum();
if (StringUtils.isNotBlank(serialNum)) {
RedisUtils.deleteObject(DEVICE_OFFLINE_WAIT_KEY_PREFIX + serialNum);
RedisUtils.deleteObject(DETECTOR_OXY_OFFLINE_WAIT_KEY_PREFIX + serialNum);
log.info("已清除设备Redis告警标记: serialNum={}", serialNum);
}
} catch (Exception e) {
log.error("清除Redis告警标记失败: deviceId={}, error={}", deviceId, e.getMessage(), e);
}
// 10. 记录操作日志
try {
MessageOpRecordUtil.addMessageOpRecordUser(
device.getUserId(), device.getUserId(),
"设备操作",
String.format("%s(%s),解除绑定。", device.getDeviceName(), device.getSerialNum())
);
} catch (Exception e) {
log.error("记录操作日志失败: deviceId={}, error={}", deviceId, e.getMessage(), e);
}
}
// 10. 最后删除设备
return baseMapper.deleteByIds(ids) > 0;
return true;
}

View File

@@ -81,7 +81,7 @@ public class DeviceWarnCombineServiceImpl implements IDeviceWarnCombineService {
.selectAll(DeviceWarnCombine.class)
.selectCount(DeviceWarnOne::getId, DeviceWarnCombineVo::getTotalCount)
.leftJoin(DeviceWarnOne.class, DeviceWarnOne::getWarnCombineId, DeviceWarnCombine::getId)
.eq(StringUtils.isNotBlank(bo.getDeviceSerialNum()), DeviceWarnCombine::getDeviceSerialNum, bo.getDeviceSerialNum())
.like(StringUtils.isNotBlank(bo.getDeviceSerialNum()), DeviceWarnCombine::getDeviceSerialNum, bo.getDeviceSerialNum())
.eq(bo.getDeviceType() != null, DeviceWarnCombine::getDeviceType, bo.getDeviceType())
.eq(bo.getThresholdType() != null, DeviceWarnCombine::getThresholdType, bo.getThresholdType())
.eq(bo.getProcessStatus() != null, DeviceWarnCombine::getProcessStatus, bo.getProcessStatus())

View File

@@ -154,6 +154,10 @@ public class LinkedCtrlServiceImpl implements ILinkedCtrlService {
.eq(bo.getOxyUpperOpen() != null, LinkedCtrl::getOxyUpperOpen, bo.getOxyUpperOpen())
.eq(bo.getOxyLowerOpen() != null, LinkedCtrl::getOxyLowerOpen, bo.getOxyLowerOpen())
.eq(bo.getIsOxyUpperTrigger() != null, LinkedCtrl::getIsOxyUpperTrigger, bo.getIsOxyUpperTrigger());
// 处理额外的查询参数
if (params != null && !params.isEmpty()) {
handleExtraParams(wrapper, params);
}
return wrapper;
}

View File

@@ -8,15 +8,12 @@ import com.intc.common.mybatis.core.page.PageQuery;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.intc.fishery.domain.*;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import com.intc.fishery.domain.bo.MessageWarnBo;
import com.intc.fishery.domain.vo.MessageWarnVo;
import com.intc.fishery.domain.MessageWarn;
import com.intc.fishery.domain.AquUser;
import com.intc.fishery.domain.Device;
import com.intc.fishery.domain.Pond;
import com.intc.fishery.mapper.MessageWarnMapper;
import com.intc.fishery.service.IMessageWarnService;
@@ -109,9 +106,11 @@ public class MessageWarnServiceImpl implements IMessageWarnService {
.eq(bo.getDeviceId() != null, MessageWarn::getDeviceId, bo.getDeviceId())
.eq(StringUtils.isNotBlank(bo.getTitle()), MessageWarn::getTitle, bo.getTitle())
.eq(StringUtils.isNotBlank(bo.getMessage()), MessageWarn::getMessage, bo.getMessage())
.like(StringUtils.isNotBlank(bo.getTitle()), MessageWarn::getTitle, bo.getTitle())
.eq(bo.getIsRead() != null, MessageWarn::getIsRead, bo.getIsRead())
.eq(bo.getWarnType() != null, MessageWarn::getWarnType, bo.getWarnType())
.like(StringUtils.isNotBlank(bo.getPondName()), MessageWarn::getPondName, bo.getPondName())
.orderByAsc(bo.getUserId() != null, MessageWarn::getIsRead)
.orderByDesc(MessageWarn::getCreateTime);
// 处理额外的查询参数
@@ -224,11 +223,11 @@ public class MessageWarnServiceImpl implements IMessageWarnService {
public Boolean readAllMessages(Long userId) {
MessageWarn update = new MessageWarn();
update.setIsRead(1);
LambdaQueryWrapper<MessageWarn> wrapper = Wrappers.lambdaQuery(MessageWarn.class)
.eq(MessageWarn::getUserId, userId)
.eq(MessageWarn::getIsRead, 0);
return baseMapper.update(update, wrapper) > 0;
}
}

View File

@@ -89,6 +89,13 @@ public class AliyunIotProperties {
*/
private int offlineWarnDelayMinutes = 6;
/**
* 太阳能网控离线告警延迟触发时间(分钟)
* 太阳能网控工作1分钟、休眠10分钟属于正常行为需要更长的延迟才能判定为真正离线
* 默认 30 分钟
*/
private int solarOfflineWarnDelayMinutes = 30;
/**
* 夜间溶解氧饱和度告警阈值(%),超过此值时开始计时
* 对应 C# DeviceSaturabilityWarnValue = 200

View File

@@ -53,4 +53,18 @@ public class AddDeviceSolarControllerBo implements Serializable {
@Schema(description = "电量电话告警下限(百分比)")
@NotNull(message = "电量电话告警下限不能为空")
private Long batteryWarnLower;
/**
* 盐度补偿值
*/
@Schema(description = "盐度补偿值")
@NotNull(message = "盐度补偿值不能为空")
private Double salinityCompensation;
/**
* 溶解氧报警下限
*/
@Schema(description = "溶解氧报警下限")
@NotNull(message = "溶解氧报警下限不能为空")
private Double oxyWarnLower;
}

View File

@@ -0,0 +1,43 @@
package com.intc.iot.domain.bo;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.constraints.NotEmpty;
import jakarta.validation.constraints.NotNull;
import lombok.Data;
import java.io.Serial;
import java.io.Serializable;
import java.util.Map;
/**
* 太阳能网控属性下发请求对象
*
* @author intc
*/
@Data
@Schema(description = "太阳能网控属性下发请求对象")
public class DeviceSolarPropertyCommandBo implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
/**
* 设备ID
*/
@Schema(description = "设备ID")
@NotNull(message = "设备ID不能为空")
private Long id;
/**
* 待下发属性key 为物模型属性标识符
*/
@Schema(description = "待下发属性")
@NotEmpty(message = "下发属性不能为空")
private Map<String, Object> properties;
/**
* 是否缓存下发。true=写入缓存设备上线后自动补发false=立即下发
*/
@Schema(description = "是否缓存下发")
private Boolean cache;
}

View File

@@ -104,6 +104,38 @@ public interface AquWarnCallNoticeMapper extends BaseMapper<AquWarnCallNotice> {
@Select("SELECT * FROM aqu_call_notice WHERE call_status = 0 AND call_time <= #{now} ORDER BY call_time ASC")
List<AquWarnCallNotice> selectDuePendingNotices(@Param("now") LocalDateTime now);
/**
* 查询指定设备在指定时间之后是否已有「等待回执中callStatus=1」或「回执成功callStatus=3」的通知记录
* 用于定时任务发出待发通知前的拦截判断:
* - callStatus=1前一条已发出但回执未到不应重复拨打等其超时后再发
* - callStatus=3已成功直接取消后续记录
*
* @param deviceId 设备ID
* @param afterTime 批次窗口起始时间
* @return 满足条件的通知数量
*/
@Select("SELECT COUNT(1) FROM aqu_call_notice " +
"WHERE device_id = #{deviceId} " +
"AND call_status IN (1, 3) " +
"AND call_time >= #{afterTime}")
long countWaitingOrSuccessByDeviceAfter(@Param("deviceId") Long deviceId,
@Param("afterTime") LocalDateTime afterTime);
/**
* 查询指定设备在指定时间之后是否已有「回执成功」的通知记录(不区分 title
* 用于定时任务发出待发通知前,判断同批次是否已有成功回执,避免重复拨打
*
* @param deviceId 设备ID
* @param afterTime 起始时间(通常取批次第一条记录的 callTime
* @return 满足条件的通知数量
*/
@Select("SELECT COUNT(1) FROM aqu_call_notice " +
"WHERE device_id = #{deviceId} " +
"AND call_status = 3 " +
"AND call_time >= #{afterTime}")
long countSuccessByDeviceAfter(@Param("deviceId") Long deviceId,
@Param("afterTime") LocalDateTime afterTime);
/**
* 通过 callNoticeId 查询关联的第一条告警消息内容
* 用于定时任务重发时构建 TTS 参数

View File

@@ -0,0 +1,34 @@
package com.intc.iot.service;
import lombok.Data;
/**
* 开关执行结果
*/
@Data
public class SwitchTurnResult {
/**
* 是否成功
*/
private boolean success;
/**
* 结果描述
*/
private String message;
public static SwitchTurnResult ok(String message) {
SwitchTurnResult result = new SwitchTurnResult();
result.setSuccess(true);
result.setMessage(message);
return result;
}
public static SwitchTurnResult fail(String message) {
SwitchTurnResult result = new SwitchTurnResult();
result.setSuccess(false);
result.setMessage(message);
return result;
}
}

View File

@@ -0,0 +1,38 @@
package com.intc.iot.service;
import com.intc.fishery.domain.TimingCtrl;
import java.util.List;
/**
* 定时控制执行服务
*/
public interface TimingCtrlExecuteService {
/**
* 按开关ID执行开/关控制
*
* @param switchId 开关ID
* @param targetStatus 目标状态1=开启0=关闭
* @param opTypeLabel 操作类型描述
* @param opMessagePrefix 操作记录前缀
* @param timingOp 是否按定时任务记录
* @return 执行结果
*/
SwitchTurnResult executeSwitch(Long switchId, Integer targetStatus, String opTypeLabel,
String opMessagePrefix, boolean timingOp);
/**
* 扫描并执行当天到点的循环定时任务
*
* @return 执行数量
*/
int scanAndExecuteDailyTimingCtrl();
/**
* 查询启用中的每天循环定时控制
*
* @return 定时控制列表
*/
List<TimingCtrl> listEnabledDailyTimingCtrls();
}

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 {
@@ -50,6 +57,9 @@ public class AmqpMessageHandlerImpl implements AmqpMessageHandler {
String topic = json.getStr("topic");
String method = json.getStr("method");
String status = json.getStr("status");
if (status != null) {
status = status.trim().toLowerCase();
}
// 检查是否是设备上线/下线消息
if (status != null && ("online".equals(status) || "offline".equals(status))) {
@@ -215,14 +225,11 @@ public class AmqpMessageHandlerImpl implements AmqpMessageHandler {
private void handleStatusMessage(JSONObject json) {
String deviceName = json.getStr("deviceName");
String status = json.getStr("status");
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);
}
/**
@@ -235,14 +242,52 @@ public class AmqpMessageHandlerImpl implements AmqpMessageHandler {
if (action == null) {
action = json.getStr("status");
}
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;
}
}

View File

@@ -2,16 +2,20 @@ 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;
import com.intc.iot.service.DeviceStatusService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.beans.factory.annotation.Autowired;
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;
/**
@@ -22,37 +26,45 @@ import java.util.Map;
*/
@Service
@RequiredArgsConstructor
@ConditionalOnBean(IotDeviceStatusMapper.class)
@Slf4j
public class DeviceStatusServiceImpl implements DeviceStatusService {
private final IotDeviceStatusMapper deviceStatusMapper;
@Autowired(required = false)
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();
@@ -62,8 +74,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 {
@@ -71,18 +87,34 @@ 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 (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);
throw e;
@@ -91,11 +123,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
@@ -121,4 +153,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;
}
}

View File

@@ -0,0 +1,363 @@
package com.intc.iot.service.impl;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.intc.common.redis.utils.RedisUtils;
import com.intc.common.tenant.helper.TenantHelper;
import com.intc.fishery.constant.DefineDeviceWarnCode;
import com.intc.fishery.domain.Device;
import com.intc.fishery.domain.DeviceSwitch;
import com.intc.fishery.domain.TimingCtrl;
import com.intc.fishery.mapper.DeviceMapper;
import com.intc.fishery.mapper.DeviceSwitchMapper;
import com.intc.fishery.mapper.TimingCtrlMapper;
import com.intc.fishery.utils.MessageOpRecordUtil;
import com.intc.iot.constant.IOTPropertyName;
import com.intc.iot.service.IotCloudService;
import com.intc.iot.service.SwitchTurnResult;
import com.intc.iot.service.TimingCtrlExecuteService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* 定时控制执行服务实现
*/
@Slf4j
@Service
@RequiredArgsConstructor
@ConditionalOnBean(IotCloudService.class)
public class TimingCtrlExecuteServiceImpl implements TimingCtrlExecuteService {
private static final String DAILY_EXEC_LOCK_PREFIX = "fishery:timingCtrl:daily:exec:";
private static final DateTimeFormatter DATE_KEY_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd");
private static final long SCAN_WINDOW_MINUTES = 4L;
/**
* 同一控制器多个开关同时到期时,每条指令之间的下发间隔(秒)
*/
private static final long SAME_DEVICE_DELAY_SECONDS = 20L;
/**
* 启动时打印日志,用于确认 Bean 是否成功创建
*/
@jakarta.annotation.PostConstruct
public void init() {
log.info("[TimingCtrl] TimingCtrlExecuteService 已初始化,扫描窗口={}分钟Redis锁前缀={}",
SCAN_WINDOW_MINUTES, DAILY_EXEC_LOCK_PREFIX);
}
@jakarta.annotation.PreDestroy
public void destroy() {
log.info("[TimingCtrl] 关闭定时控制指令调度线程池...");
commandScheduler.shutdown();
try {
if (!commandScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
commandScheduler.shutdownNow();
}
} catch (InterruptedException e) {
commandScheduler.shutdownNow();
Thread.currentThread().interrupt();
}
}
private final DeviceMapper deviceMapper;
private final DeviceSwitchMapper deviceSwitchMapper;
private final TimingCtrlMapper timingCtrlMapper;
private final IotCloudService iotCloudService;
/**
* 延迟下发指令的调度线程池(守护线程,不阻塞应用关闭)
*/
private final ScheduledExecutorService commandScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
Thread t = new Thread(r, "timing-ctrl-cmd-scheduler");
t.setDaemon(true);
return t;
});
@Override
public SwitchTurnResult executeSwitch(Long switchId, Integer targetStatus, String opTypeLabel,
String opMessagePrefix, boolean timingOp) {
try {
if (switchId == null || targetStatus == null) {
return SwitchTurnResult.fail("参数不能为空");
}
DeviceSwitch deviceSwitch = deviceSwitchMapper.selectOne(
new LambdaQueryWrapper<DeviceSwitch>()
.eq(DeviceSwitch::getId, switchId)
.select(DeviceSwitch::getId, DeviceSwitch::getDeviceId, DeviceSwitch::getIndex,
DeviceSwitch::getSwitchName, DeviceSwitch::getIsOpen, DeviceSwitch::getLastTurnTime)
);
if (deviceSwitch == null) {
return SwitchTurnResult.fail("开关不存在");
}
if (deviceSwitch.getLastTurnTime() != null) {
long timeDiff = System.currentTimeMillis() - deviceSwitch.getLastTurnTime().getTime();
if (timeDiff < 3000) {
return SwitchTurnResult.fail("操作过于频繁,请稍后再试");
}
}
Device device = deviceMapper.selectOne(
new LambdaQueryWrapper<Device>()
.eq(Device::getId, deviceSwitch.getDeviceId())
.select(Device::getId, Device::getUserId, Device::getIotId, Device::getSerialNum,
Device::getDeviceName, Device::getWarnCode, Device::getDeadTime)
);
if (device == null) {
return SwitchTurnResult.fail("开关不存在或无权限访问");
}
if (device.getDeadTime() != null && new Date().after(device.getDeadTime())) {
return SwitchTurnResult.fail("设备服务已过期");
}
if (device.getWarnCode() == null || device.getWarnCode() == 99
|| (device.getWarnCode() & DefineDeviceWarnCode.DeviceOffline) != 0) {
return SwitchTurnResult.fail("设备离线或已关机");
}
if (iotCloudService == null) {
return SwitchTurnResult.fail("物联网服务未启用");
}
Map<String, Object> properties = new HashMap<>();
properties.put(IOTPropertyName.SWITCH_INDEX + deviceSwitch.getIndex(), targetStatus);
boolean success = iotCloudService.setProperty(device.getIotId(), properties, true, 2);
if (!success) {
return SwitchTurnResult.fail("设置开关失败");
}
Date now = new Date();
deviceSwitchMapper.update(null,
new com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper<DeviceSwitch>()
.eq(DeviceSwitch::getId, switchId)
.set(DeviceSwitch::getIsOpen, targetStatus)
.set(DeviceSwitch::getLastTurnTime, now)
);
if (device.getUserId() != null && opTypeLabel != null && timingOp) {
String deviceLabel = device.getDeviceName() != null ? device.getDeviceName() : device.getSerialNum();
String switchLabel = deviceSwitch.getSwitchName() != null
? deviceSwitch.getSwitchName()
: ("Switch" + deviceSwitch.getIndex());
String operation = targetStatus == 1 ? "开启" : "关闭";
String message = String.format("%s(%s)的开关%s%s%s。",
deviceLabel, device.getSerialNum(), switchLabel, operation,
opMessagePrefix != null ? "" + opMessagePrefix + "" : "");
MessageOpRecordUtil.addMessageOpRecordTimingLoop(device.getUserId(), opTypeLabel, message);
}
log.info("[SwitchTurn] 设备={} 开关{} {}成功",
device.getSerialNum() != null ? device.getSerialNum() : device.getIotId(),
deviceSwitch.getSwitchName(),
targetStatus == 1 ? "开启" : "关闭");
return SwitchTurnResult.ok("操作成功");
} catch (Exception e) {
log.error("[SwitchTurn] 开关执行失败: {}", e.getMessage(), e);
return SwitchTurnResult.fail("开关控制失败: " + e.getMessage());
}
}
@Override
public int scanAndExecuteDailyTimingCtrl() {
// 后台定时任务没有租户上下文必须忽略租户过滤否则SQL会拼接 WHERE tenant_id = NULL 导致查不到数据
return TenantHelper.ignore(() -> doScanAndExecuteDailyTimingCtrl());
}
private int doScanAndExecuteDailyTimingCtrl() {
List<TimingCtrl> list = listEnabledDailyTimingCtrls();
if (list.isEmpty()) {
log.info("[TimingCtrl] 数据库中没有启用的每天循环定时控制记录(isOpen=1, loopType=1)");
return 0;
}
LocalDateTime now = LocalDateTime.now();
log.info("[TimingCtrl] 扫描开始,当前时间={},查到 {} 条启用的定时控制", now, list.size());
// 第一轮收集所有到期任务按设备ID分组
Map<Long, List<DueTask>> tasksByDevice = new LinkedHashMap<>();
for (TimingCtrl timingCtrl : list) {
try {
if (timingCtrl.getSwitchId() == null || timingCtrl.getOpenTime() == null || timingCtrl.getCloseTime() == null) {
log.debug("[TimingCtrl] 跳过: 字段为空id={}, switchId={}, openTime={}, closeTime={}",
timingCtrl.getId(), timingCtrl.getSwitchId(), timingCtrl.getOpenTime(), timingCtrl.getCloseTime());
continue;
}
LocalTime openTime = toLocalTime(timingCtrl.getOpenTime());
LocalTime closeTime = toLocalTime(timingCtrl.getCloseTime());
boolean openDue = isDueInLookbackWindow(now, openTime);
boolean closeDue = isDueInLookbackWindow(now, closeTime);
if (!openDue && !closeDue) {
log.debug("[TimingCtrl] 跳过: 未到执行时间id={}, openTime={}, closeTime={}, now={}",
timingCtrl.getId(), openTime, closeTime, now.toLocalTime());
continue;
}
// 查询开关所属设备ID用于按设备分组错峰下发
Long deviceId = getDeviceIdBySwitchId(timingCtrl.getSwitchId());
if (deviceId == null) {
log.warn("[TimingCtrl] 开关不存在跳过switchId={}", timingCtrl.getSwitchId());
continue;
}
if (openDue) {
if (tryLock(timingCtrl.getId(), now.toLocalDate(), "open")) {
tasksByDevice.computeIfAbsent(deviceId, k -> new ArrayList<>())
.add(new DueTask(timingCtrl.getId(), timingCtrl.getSwitchId(), 1));
} else {
log.debug("[TimingCtrl] 开启: 今天已执行过跳过id={}", timingCtrl.getId());
}
}
if (closeDue) {
if (tryLock(timingCtrl.getId(), now.toLocalDate(), "close")) {
tasksByDevice.computeIfAbsent(deviceId, k -> new ArrayList<>())
.add(new DueTask(timingCtrl.getId(), timingCtrl.getSwitchId(), 0));
} else {
log.debug("[TimingCtrl] 关闭: 今天已执行过跳过id={}", timingCtrl.getId());
}
}
} catch (Exception e) {
log.error("[TimingCtrl] 处理定时控制失败id={}: {}", timingCtrl.getId(), e.getMessage(), e);
}
}
if (tasksByDevice.isEmpty()) {
log.info("[TimingCtrl] 本次扫描没有需要执行的定时控制");
return 0;
}
// 第二轮:按设备分组下发指令,同一设备的多个开关每隔 SAME_DEVICE_DELAY_SECONDS 秒下发一条
int executedCount = 0;
for (Map.Entry<Long, List<DueTask>> entry : tasksByDevice.entrySet()) {
Long deviceId = entry.getKey();
List<DueTask> tasks = entry.getValue();
log.info("[TimingCtrl] 设备{} 共有 {} 条定时控制待执行", deviceId, tasks.size());
for (int i = 0; i < tasks.size(); i++) {
DueTask task = tasks.get(i);
long delaySeconds = i * SAME_DEVICE_DELAY_SECONDS;
if (delaySeconds == 0) {
// 第一条立即下发
SwitchTurnResult result = executeSwitch(task.switchId, task.targetStatus,
"定时控制管理", "定时任务到达" + (task.targetStatus == 1 ? "开启" : "关闭") + "时间", true);
if (result.isSuccess()) {
executedCount++;
} else {
log.warn("[TimingCtrl] 执行失败timingCtrlId={}, msg={}", task.timingCtrlId, result.getMessage());
}
} else {
// 同一设备后续开关延迟下发,避免短时间内集中下发多条指令导致硬件无法及时处理
final Long tcId = task.timingCtrlId;
final Long switchId = task.switchId;
final int targetStatus = task.targetStatus;
final long delay = delaySeconds;
log.info("[TimingCtrl] 定时控制{} 将延迟{}秒后执行(设备{}排队)", tcId, delay, deviceId);
commandScheduler.schedule(() -> {
try {
TenantHelper.ignore(() -> {
log.info("[TimingCtrl] 开始延迟执行定时控制timingCtrlId={}", tcId);
SwitchTurnResult result = executeSwitch(switchId, targetStatus,
"定时控制管理", "定时任务到达" + (targetStatus == 1 ? "开启" : "关闭") + "时间", true);
if (result.isSuccess()) {
log.info("[TimingCtrl] 延迟执行成功timingCtrlId={}", tcId);
} else {
log.warn("[TimingCtrl] 延迟执行失败timingCtrlId={}, msg={}", tcId, result.getMessage());
}
});
} catch (Exception e) {
log.error("[TimingCtrl] 延迟执行异常timingCtrlId={}: {}", tcId, e.getMessage(), e);
}
}, delay, TimeUnit.SECONDS);
executedCount++;
}
}
}
return executedCount;
}
/**
* 根据开关ID查询所属设备ID
*/
private Long getDeviceIdBySwitchId(Long switchId) {
DeviceSwitch ds = deviceSwitchMapper.selectOne(
new LambdaQueryWrapper<DeviceSwitch>()
.eq(DeviceSwitch::getId, switchId)
.select(DeviceSwitch::getDeviceId)
);
return ds != null ? ds.getDeviceId() : null;
}
/**
* 到期的定时控制任务
*/
private static class DueTask {
final Long timingCtrlId;
final Long switchId;
final int targetStatus;
DueTask(Long timingCtrlId, Long switchId, int targetStatus) {
this.timingCtrlId = timingCtrlId;
this.switchId = switchId;
this.targetStatus = targetStatus;
}
}
@Override
public List<TimingCtrl> listEnabledDailyTimingCtrls() {
return timingCtrlMapper.selectList(
new LambdaQueryWrapper<TimingCtrl>()
.eq(TimingCtrl::getIsOpen, 1L)
.eq(TimingCtrl::getLoopType, 2L)
.select(TimingCtrl::getId, TimingCtrl::getSwitchId, TimingCtrl::getOpenTime,
TimingCtrl::getCloseTime, TimingCtrl::getLoopType, TimingCtrl::getIsOpen)
);
}
private static LocalTime toLocalTime(Date date) {
return LocalDateTime.ofInstant(date.toInstant(), java.time.ZoneId.systemDefault()).toLocalTime();
}
private static boolean isDueInLookbackWindow(LocalDateTime now, LocalTime scheduleTime) {
LocalDateTime scheduleDateTime = LocalDateTime.of(now.toLocalDate(), scheduleTime);
if (scheduleDateTime.isAfter(now)) {
scheduleDateTime = scheduleDateTime.minusDays(1);
}
LocalDateTime lowerBound = now.minusMinutes(SCAN_WINDOW_MINUTES);
return !scheduleDateTime.isBefore(lowerBound) && !scheduleDateTime.isAfter(now);
}
private boolean tryLock(Long timingCtrlId, LocalDate date, String action) {
String redisKey = DAILY_EXEC_LOCK_PREFIX + timingCtrlId + ":" + date.format(DATE_KEY_FORMATTER) + ":" + action;
boolean locked = RedisUtils.setObjectIfAbsent(redisKey, "1", Duration.ofHours(25));
if (locked) {
log.info("[TimingCtrl] 获取执行锁成功timingCtrlId={}, date={}, action={}", timingCtrlId, date, action);
}
return locked;
}
}

View File

@@ -50,8 +50,8 @@ public class VmsMnsConsumerServiceImpl implements VmsMnsConsumerService {
/** MNS 消息类型(固定值) */
private static final String VMS_MNS_MESSAGE_TYPE = "VoiceReport";
/** MNS 默认线程数 */
private static final int MNS_THREAD_COUNT = 2;
/** MNS 默认线程数改为1避免多线程共享 mnsClient 时互相踢连接导致无限崩溃 */
private static final int MNS_THREAD_COUNT = 1;
private static final int POLL_WAIT_SECONDS = 30;
/** Token 过期缓冲时间120秒小于此值则重新获取防止服务器时间差异 */

View File

@@ -43,13 +43,21 @@ public class VmsNoticeServiceImpl implements VmsNoticeService {
if (vmsClient == null) {
synchronized (this) {
if (vmsClient == null) {
// 必须使用 VMS 专用 AKvms.access-key-id而非 IoT 平台 AK
// MNS 回执消费也使用 VMS 专用 AK两端必须一致否则回执无法匹配
AliyunIotProperties.VmsMnsConfig vmsCfg = aliyunIotProperties.getVms();
String akId = (vmsCfg.getAccessKeyId() != null && !vmsCfg.getAccessKeyId().isEmpty())
? vmsCfg.getAccessKeyId() : aliyunIotProperties.getAccessKeyId();
String akSecret = (vmsCfg.getAccessKeySecret() != null && !vmsCfg.getAccessKeySecret().isEmpty())
? vmsCfg.getAccessKeySecret() : aliyunIotProperties.getAccessKeySecret();
// VMS 服务固定在 cn-hangzhou与 IoT 平台 region 无关
DefaultProfile profile = DefaultProfile.getProfile(
"cn-hangzhou",
aliyunIotProperties.getAccessKeyId(),
aliyunIotProperties.getAccessKeySecret()
akId,
akSecret
);
vmsClient = new DefaultAcsClient(profile);
log.info("VMS 客户端初始化成功");
log.info("VMS 客户端初始化成功, akId={}", akId);
}
}
}

View File

@@ -56,11 +56,22 @@ public class WarnCallNoticeServiceImpl implements WarnCallNoticeService {
/**
* 回执超时时间(分钟)
* VMS 回执延迟较长,设为 15 分钟,给 VMS 平台充足的回调缓冲时间
* 注意:超时判断基于 callTime预计呼叫时间实际呼叫发起后回调可能延迟数分钟到达
* 从呼叫实际发出callTime 已修正为真实发出时间)到收到 VMS 回执的最长等待时间
* 正常链路:通话约 30s + VMS 生成回执约 1~2min通常 3 分钟内完成。
* 但阿里云 VMS 在网络波动或平台繁忙时,回执延迟可达 5~10 分钟,因此设为 15 分钟。
* 这意味着同批次最多等待 15 分钟才会解除阻塞让下一条记录发出,
* 整体上 15 分钟内不会对同一设备重复拨打,是可接受的保守策略。
*/
private static final int CALLBACK_TIMEOUT_MINUTES = 15;
/**
* 批次阻断查询窗口(分钟)
* 用于定时任务发出待发记录前,向前扫描同设备是否已有处于这个窗口内的阻塞记录。
* 计算依据:回执超时(15min) + 批次最大时长(3次重试 x 2个手机号 x 3min = 18min) + 缓冲 = 40min
* 确保给定批次的所有记录都在窗口内,防止批次尾部记录因窗口过短而脱革导致重复发出。
*/
private static final int BATCH_BLOCK_WINDOW_MINUTES = 40;
@Override
@Transactional(rollbackFor = Exception.class)
public boolean processVmsCallback(VmsCallback callback) {
@@ -258,6 +269,20 @@ public class WarnCallNoticeServiceImpl implements WarnCallNoticeService {
int sentCount = 0;
for (AquWarnCallNotice notice : pendingList) {
try {
// 发出前先检查同设备同批次窗口内的状态
// 批次窗口 = 当前记录的 callTime 往前 BATCH_BLOCK_WINDOW_MINUTES 分钟(足够覆盖整个批次)
// - status=1等待回执中前一条已发出但回执未到不重复拨打等其超时后再发
// - status=3已成功已有成功回执跳过该条记录
if (notice.getDeviceId() != null) {
LocalDateTime batchWindowStart = notice.getCallTime().minusMinutes(BATCH_BLOCK_WINDOW_MINUTES);
long blockingCount = warnCallNoticeMapper.countWaitingOrSuccessByDeviceAfter(
notice.getDeviceId(), batchWindowStart);
if (blockingCount > 0) {
log.info("[待发通知] 设备 {} 同批次内已有等待回执或成功的记录跳过记录ID: {}", notice.getDeviceId(), notice.getId());
continue;
}
}
// 查询关联的告警标题title作为 TTS warnMessage 参数
// 使用 title如"设备离线"、"溶解氧")而非完整 message避免长字符串被 TTS 逐字播报
String warnTitle = warnCallNoticeMapper.selectWarnTitleByCallNoticeId(notice.getId());
@@ -291,6 +316,7 @@ public class WarnCallNoticeServiceImpl implements WarnCallNoticeService {
notice.setCallStatus(CALL_STATUS_CALL_SUCCESS);
notice.setCallId(response.getCallId());
notice.setOutId(outId);
notice.setCallTime(LocalDateTime.now()); // 记录实际发出时间,用于超时判断
warnCallNoticeMapper.updateById(notice);
sentCount++;
log.info("[待发通知] 呼叫成功: {} - 记录ID: {}", notice.getMobilePhone(), notice.getId());

View File

@@ -0,0 +1,83 @@
package com.intc.iot.task;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.intc.fishery.domain.Device;
import com.intc.fishery.mapper.DeviceMapper;
import com.intc.iot.handler.DeviceDataHandler;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.time.LocalTime;
import java.util.Date;
import java.util.List;
/**
* 设备即将到期告警定时任务
* 每小时扫描一次设备表对到期时间距离当前不到24小时的设备触发电话通知提醒用户续费。
*
* @author intc-iot
*/
@Slf4j
@Component
@RequiredArgsConstructor
@ConditionalOnBean(DeviceDataHandler.class)
public class DeviceExpirationWarnTask {
private final DeviceDataHandler deviceDataHandler;
private final DeviceMapper deviceMapper;
/**
* 每2小时执行一次扫描即将到期24小时内的设备并触发电话通知
*/
@Scheduled(fixedDelay = 7200000)
public void checkDeviceExpirationAndNotify() {
try {
// 只在 8:00 ~ 21:00 之间拨打电话通知,避免夜间打扰用户
LocalTime currentTime = LocalTime.now();
if (currentTime.isBefore(LocalTime.of(8, 0)) || !currentTime.isBefore(LocalTime.of(21, 0))) {
log.debug("[到期告警] 当前时间 {} 不在电话通知时段(08:00~21:00),跳过本次扫描", currentTime);
return;
}
Date now = new Date();
// 24小时后的时间点
Date deadline = new Date(now.getTime() + 24L * 60 * 60 * 1000);
// 查询所有满足条件的设备:
// 1. deadTime 不为空
// 2. deadTime > 当前时间(尚未到期)
// 3. deadTime <= 当前时间 + 24小时即将到期
// 4. userId 不为空(已绑定用户)
List<Device> expiringDevices = deviceMapper.selectList(
new LambdaQueryWrapper<Device>()
.isNotNull(Device::getDeadTime)
.gt(Device::getDeadTime, now)
.le(Device::getDeadTime, deadline)
.isNotNull(Device::getUserId)
.select(Device::getId, Device::getUserId, Device::getSerialNum,
Device::getDeviceName, Device::getDeviceType, Device::getPondId,
Device::getDeadTime)
);
if (expiringDevices == null || expiringDevices.isEmpty()) {
return;
}
log.info("[到期告警] 扫描到 {} 台即将到期的设备,开始逐一通知", expiringDevices.size());
for (Device device : expiringDevices) {
try {
deviceDataHandler.triggerDeviceExpirationAlarm(device);
} catch (Exception e) {
log.error("[到期告警] 处理单台设备异常deviceId={}: {}", device.getId(), e.getMessage(), e);
}
}
} catch (Exception e) {
log.error("[到期告警] 扫描设备到期告警异常: {}", e.getMessage(), e);
}
}
}

View File

@@ -1,6 +1,7 @@
package com.intc.iot.task;
import com.intc.common.redis.utils.RedisUtils;
import com.intc.fishery.constant.DefineDeviceWarnCode;
import com.intc.iot.config.AliyunIotProperties;
import com.intc.iot.handler.DeviceDataHandler;
import lombok.RequiredArgsConstructor;
@@ -28,6 +29,12 @@ public class DeviceOfflineWarnTask {
private static final String OFFLINE_WAIT_KEY_PREFIX = "device:offline:wait:";
/** Redis value 分隔符,格式:{offlineTime}|{deviceType} */
private static final String VALUE_SEPARATOR = "|";
/** 太阳能网控设备类型值 */
private static final int SOLAR_CONTROLLER_TYPE = 3;
private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
private final DeviceDataHandler deviceDataHandler;
@@ -36,7 +43,8 @@ public class DeviceOfflineWarnTask {
/**
* 每分钟扫描一次离线等待标记
* 对超过配置的 offline-warn-delay-minutes 分钟的设备触发告警
* 对超过配置的延迟时长的设备触发告警
* Redis value 格式:{offlineTime}|{deviceType},根据设备类型使用对应延迟时长
*/
@Scheduled(fixedDelay = 60000)
public void checkOfflineWaitAndTrigger() {
@@ -49,22 +57,48 @@ public class DeviceOfflineWarnTask {
LocalDateTime now = LocalDateTime.now();
for (String key : keys) {
try {
String offlineTimeStr = RedisUtils.getCacheObject(key);
if (offlineTimeStr == null) {
String value = RedisUtils.getCacheObject(key);
if (value == null) {
continue;
}
// 解析 value兼容旧格式纯时间字符串和新格式时间|设备类型)
String offlineTimeStr;
int deviceType = 0;
if (value.contains(VALUE_SEPARATOR)) {
String[] parts = value.split("\\" + VALUE_SEPARATOR, 2);
offlineTimeStr = parts[0];
try {
deviceType = Integer.parseInt(parts[1]);
} catch (NumberFormatException ignore) {
}
} else {
// 旧格式兼容:纯时间字符串,设备类型未知,使用默认延迟
offlineTimeStr = value;
}
LocalDateTime offlineTime = LocalDateTime.parse(offlineTimeStr, FORMATTER);
long elapsedMinutes = java.time.Duration.between(offlineTime, now).toMinutes();
if (elapsedMinutes >= aliyunIotProperties.getOfflineWarnDelayMinutes()) {
// 根据设备类型选择对应的延迟时长
int delayMinutes = (deviceType == SOLAR_CONTROLLER_TYPE)
? aliyunIotProperties.getSolarOfflineWarnDelayMinutes()
: aliyunIotProperties.getOfflineWarnDelayMinutes();
if (elapsedMinutes >= delayMinutes) {
// 提取 deviceName去掉前缀
String deviceName = key.substring(OFFLINE_WAIT_KEY_PREFIX.length());
// 触发告警前先删除标记,防止重复触发
RedisUtils.deleteObject(key);
log.info("[离线告警] 设备离线已超过 {} 分钟,触发告警: {}", elapsedMinutes, deviceName);
log.info("[离线告警] 设备离线已超过 {} 分钟(阈值{}分钟),触发告警: {}", elapsedMinutes, delayMinutes, deviceName);
// 太阳能网控设备超时后,从休眠状态转为真正离线:置 DeviceOffline 位,清 DeviceSleeping 位
if (deviceType == SOLAR_CONTROLLER_TYPE) {
deviceDataHandler.setDeviceOfflineAndClearSleeping(deviceName);
}
deviceDataHandler.triggerOfflineAlarm(deviceName);
}
} catch (Exception e) {
@@ -75,4 +109,50 @@ public class DeviceOfflineWarnTask {
log.error("[离线告警] 扫描离线标记异常: {}", e.getMessage(), e);
}
}
/**
* 每分钟扫描一次溶解氧探头离线等待标记
* 对超过配置的延迟时长的设备触发探头离线告警
* Redis value 格式:{offlineTime}yyyy-MM-dd HH:mm:ss
*/
@Scheduled(fixedDelay = 60000)
public void checkDetectorOxyOfflineWaitAndTrigger() {
try {
Collection<String> keys = RedisUtils.keys(DeviceDataHandler.DETECTOR_OXY_OFFLINE_WAIT_KEY_PREFIX + "*");
if (keys == null || keys.isEmpty()) {
return;
}
LocalDateTime now = LocalDateTime.now();
int delayMinutes = aliyunIotProperties.getOfflineWarnDelayMinutes();
for (String key : keys) {
try {
String offlineTimeStr = RedisUtils.getCacheObject(key);
if (offlineTimeStr == null) {
continue;
}
LocalDateTime offlineTime = LocalDateTime.parse(offlineTimeStr, FORMATTER);
long elapsedMinutes = java.time.Duration.between(offlineTime, now).toMinutes();
if (elapsedMinutes >= delayMinutes) {
// 提取 deviceName去掉前缀
String deviceName = key.substring(DeviceDataHandler.DETECTOR_OXY_OFFLINE_WAIT_KEY_PREFIX.length());
// 触发告警前先删除标记,防止重复触发
RedisUtils.deleteObject(key);
log.info("[探头离线告警] 溶解氧探头离线已超过 {} 分钟(阈值{}分钟),触发告警: {}", elapsedMinutes, delayMinutes, deviceName);
deviceDataHandler.triggerDetectorOxyOfflineAlarm(deviceName);
}
} catch (Exception e) {
log.error("[探头离线告警] 处理单条探头离线标记异常key={}: {}", key, e.getMessage(), e);
}
}
} catch (Exception e) {
log.error("[探头离线告警] 扫描探头离线标记异常: {}", e.getMessage(), e);
}
}
}

View File

@@ -0,0 +1,38 @@
package com.intc.iot.task;
import com.intc.iot.service.TimingCtrlExecuteService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
/**
* 开关定时任务执行扫描
*/
@Slf4j
@Component
@RequiredArgsConstructor
@ConditionalOnBean(TimingCtrlExecuteService.class)
public class TimingCtrlExecuteTask {
private static final long SCAN_DELAY_MS = 120000L;
private static final long INITIAL_DELAY_MS = 10000L;
private final TimingCtrlExecuteService timingCtrlExecuteService;
/**
* 每5分钟扫描一次启用中的每天循环定时控制
* 启动后10秒执行第一次扫描
*/
@Scheduled(fixedDelay = SCAN_DELAY_MS, initialDelay = INITIAL_DELAY_MS)
public void scanDailyTimingCtrl() {
try {
log.info("[TimingCtrlTask] 开始扫描定时控制...");
int count = timingCtrlExecuteService.scanAndExecuteDailyTimingCtrl();
log.info("[TimingCtrlTask] 本次扫描完成,执行了 {} 条定时控制", count);
} catch (Exception e) {
log.error("[TimingCtrlTask] 扫描定时控制失败: {}", e.getMessage(), e);
}
}
}

View File

@@ -0,0 +1,29 @@
package com.intc.tdengine.mapper;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.Date;
/**
* 设备到期信息Mapper查询MySQL主库
*
* @author intc
*/
@Repository
@Mapper
public interface DeviceExpireMapper {
/**
* 根据设备序列号查询服务到期时间
*
* @param serialNum 设备序列号
* @return 服务到期时间设备不存在时返回null
*/
@DS("master")
@InterceptorIgnore(tenantLine = "true")
Date getDeadTimeBySerialNum(@Param("serialNum") String serialNum);
}

View File

@@ -77,4 +77,23 @@ public interface DeviceSensorDataMapper {
@Param("endTime") String endTime,
@Param("intervalType") Integer intervalType);
/**
* 查询设备最新上数时间
*
* @param serialNum 设备序列号
* @return 最新传感器数据
*/
@DS("taos")
@InterceptorIgnore(tenantLine = "true")
DeviceSensorData getLatestReportTime(@Param("serialNum") String serialNum);
/**
* 删除设备子表(清除该设备的所有时序数据)
*
* @param serialNum 设备序列号
*/
@DS("taos")
@InterceptorIgnore(tenantLine = "true")
void dropSubTable(@Param("serialNum") String serialNum);
}

View File

@@ -38,4 +38,19 @@ public interface IDeviceSensorDataService {
* @return 设备传感器数据列表
*/
public List<DeviceSensorData> getHistoryDataList(String serialNum, String startTime, String endTime, Integer intervalType);
/**
* 查询设备最新上数时间
*
* @param serialNum 设备序列号
* @return 最新上数时间字符串,格式 yyyy-MM-dd HH:mm:ss无数据时返回 null
*/
String getLatestReportTime(String serialNum);
/**
* 删除设备子表(清除该设备的所有时序数据)
*
* @param serialNum 设备序列号
*/
void dropSubTable(String serialNum);
}

View File

@@ -2,6 +2,7 @@ package com.intc.tdengine.service.impl;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.intc.tdengine.domain.DeviceSensorData;
import com.intc.tdengine.mapper.DeviceExpireMapper;
import com.intc.tdengine.mapper.DeviceSensorDataMapper;
import com.intc.tdengine.service.IDeviceSensorDataService;
import jakarta.annotation.Resource;
@@ -9,6 +10,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
@Slf4j
@@ -17,6 +19,9 @@ import java.util.List;
public class DeviceSensorDataService implements IDeviceSensorDataService {
@Resource
private DeviceSensorDataMapper deviceSensorDataMapper;
@Resource
private DeviceExpireMapper deviceExpireMapper;
/**
* 已创建的子表缓存(设备序列号)
@@ -134,33 +139,36 @@ public class DeviceSensorDataService implements IDeviceSensorDataService {
}
try {
// 打印第一条数据的详细信息
if (!dataList.isEmpty()) {
DeviceSensorData firstData = dataList.get(0);
// 检查子表是否存在(使用缓存优化)
String serialNum = firstData.getSerialNum();
// 按 serialNum 分组,对每个设备单独检查并确保子表存在
java.util.Map<String, DeviceSensorData> firstBySerial = new java.util.LinkedHashMap<>();
for (DeviceSensorData data : dataList) {
String sn = data.getSerialNum();
if (sn != null && !sn.isEmpty() && !firstBySerial.containsKey(sn)) {
firstBySerial.put(sn, data);
}
}
for (java.util.Map.Entry<String, DeviceSensorData> entry : firstBySerial.entrySet()) {
String serialNum = entry.getKey();
String tableName = "t_" + serialNum;
// 先检查缓存,避免重复查询数据库
if (!CREATED_TABLES.containsKey(serialNum)) {
try {
Integer exists = deviceSensorDataMapper.checkTableExists(serialNum);
if (exists == null || exists == 0) {
deviceSensorDataMapper.createTable(firstData);
deviceSensorDataMapper.createTable(entry.getValue());
log.debug("子表 {} 已创建", tableName);
}
// 加入缓存
CREATED_TABLES.put(serialNum, true);
} catch (Exception e) {
// 检查表不存在是正常情况,使用 USING 语法会自动创建
String errorMsg = e.getMessage();
if (errorMsg != null && errorMsg.contains("Table does not exist")) {
log.debug("子表 {} 不存在,将自动创建", tableName);
} else {
log.warn("检查子表失败: {}", errorMsg != null && errorMsg.length() > 100 ? errorMsg.substring(0, 100) : errorMsg);
log.warn("检查子表失败[{}]: {}", tableName,
errorMsg != null && errorMsg.length() > 100 ? errorMsg.substring(0, 100) : errorMsg);
}
// 即使失败也加入缓存,避免重复尝
CREATED_TABLES.put(serialNum, true);
// 建表异常时不加入缓存,下次重
}
} else {
log.debug("子表 {} 已在缓存中", tableName);
@@ -203,6 +211,13 @@ public class DeviceSensorDataService implements IDeviceSensorDataService {
public List<DeviceSensorData> getHistoryDataList(String serialNum, String startTime, String endTime, Integer intervalType) {
List<DeviceSensorData> list = new ArrayList<>();
try {
// 检查设备是否已过期
Date deadTime = deviceExpireMapper.getDeadTimeBySerialNum(serialNum);
if (deadTime != null && deadTime.before(new Date())) {
log.info("设备已过期,不返回历史数据: serialNum={}, deadTime={}", serialNum, deadTime);
return list;
}
// 默认为原始数据
if (intervalType == null) {
intervalType = 1;
@@ -241,6 +256,31 @@ public class DeviceSensorDataService implements IDeviceSensorDataService {
}
return list;
}
/**
* 查询设备最新上数时间
*/
@Override
public String getLatestReportTime(String serialNum) {
if (serialNum == null || serialNum.isBlank()) {
return null;
}
try {
DeviceSensorData latest = deviceSensorDataMapper.getLatestReportTime(serialNum);
if (latest == null) {
return null;
}
return latest.getCreateTime() != null ? latest.getCreateTime() : latest.getTime();
} catch (Exception e) {
String errorMsg = e.getMessage();
if (errorMsg != null && errorMsg.toLowerCase().contains("not exist")) {
log.debug("设备TDengine子表不存在或无数据: serialNum={}", serialNum);
} else {
log.warn("查询设备最新上数时间失败: serialNum={}, error={}", serialNum, errorMsg);
}
return null;
}
}
/**
* 获取间隔类型描述
@@ -256,4 +296,27 @@ public class DeviceSensorDataService implements IDeviceSensorDataService {
default: return "未知";
}
}
/**
* 删除设备子表(清除该设备的所有时序数据)
*/
@Override
public void dropSubTable(String serialNum) {
if (serialNum == null || serialNum.isBlank()) {
return;
}
try {
deviceSensorDataMapper.dropSubTable(serialNum);
// 清除建表缓存,下次插入数据时会重新建表
CREATED_TABLES.remove(serialNum);
log.info("已删除设备TDengine子表: serialNum={}", serialNum);
} catch (Exception e) {
String errorMsg = e.getMessage();
if (errorMsg != null && errorMsg.toLowerCase().contains("not exist")) {
log.debug("设备TDengine子表不存在无需删除: serialNum={}", serialNum);
} else {
log.warn("删除设备TDengine子表失败: serialNum={}, error={}", serialNum, errorMsg);
}
}
}
}

View File

@@ -0,0 +1,10 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.intc.tdengine.mapper.DeviceExpireMapper">
<select id="getDeadTimeBySerialNum" resultType="java.util.Date">
SELECT dead_time FROM aqu_device WHERE serial_num = #{serialNum} LIMIT 1
</select>
</mapper>

View File

@@ -111,4 +111,17 @@
</choose>
</if>
</select>
<select id="getLatestReportTime" resultType="com.intc.tdengine.domain.DeviceSensorData">
SELECT
`time`,
createTime
FROM `fishery`.`t_${serialNum}`
ORDER BY `time` DESC
LIMIT 1
</select>
<update id="dropSubTable">
DROP TABLE IF EXISTS `fishery`.`t_${serialNum}`
</update>
</mapper>

View File

@@ -262,6 +262,7 @@ public class WeixinController extends BaseController {
// 9. 构建返回结果
WxPayOrderVo result = new WxPayOrderVo();
result.setPrepayId(prepayId);
result.setOutTradeNumber(order.getOutTradeNumber());
result.setTotalAmount(order.getTotalAmount());
result.setTimestamp(payParams.get("timeStamp"));
result.setNonceStr(payParams.get("nonceStr"));
@@ -278,6 +279,36 @@ public class WeixinController extends BaseController {
}
}
@Operation(summary = "主动确认微信支付订单")
@PostMapping("/pay/confirm_order")
public R<Boolean> confirmOrder(@RequestBody Map<String, Object> request) {
try {
if (payOrderBusinessService == null) {
return R.fail("支付服务未启用");
}
String outTradeNumber = request != null && request.get("outTradeNumber") != null
? String.valueOf(request.get("outTradeNumber"))
: null;
if ((outTradeNumber == null || outTradeNumber.isEmpty()) && request != null && request.get("orderId") != null) {
Long orderId = Long.valueOf(String.valueOf(request.get("orderId")));
com.intc.fishery.domain.PayOrder order = payOrderBusinessService.queryById(orderId);
if (order != null) {
outTradeNumber = order.getOutTradeNumber();
}
}
if (outTradeNumber == null || outTradeNumber.isEmpty()) {
return R.fail("商户订单号不能为空");
}
boolean success = payOrderBusinessService.confirmPaidOrder(outTradeNumber);
return success ? R.ok(true) : R.fail("订单未支付成功或确认失败");
} catch (Exception e) {
log.error("确认支付订单失败", e);
return R.fail("确认支付订单失败: " + e.getMessage());
}
}
@Operation(summary = "微信支付回调通知")
@PostMapping("/pay_notify")
public Map<String, String> payNotify(

View File

@@ -24,6 +24,12 @@ public class WxPayOrderVo implements Serializable {
@Schema(description = "预支付ID")
private String prepayId;
/**
* 商户系统内部订单号
*/
@Schema(description = "商户订单号")
private String outTradeNumber;
/**
* 签名类型
*/

View File

@@ -82,4 +82,28 @@ public interface PayOrderBusinessService {
* @return 是否更新成功
*/
boolean updatePrepayId(Long orderId, String prepayId);
/**
* 补偿未处理的支付订单
* 主动向微信支付查询状态为“未支付/支付中/支付失败/已关闭”且有商户订单号的订单,
* 如果微信侧已支付成功则执行回调补偿
*
* @return 补偿成功的订单数量
*/
int compensateUnpaidOrders();
/**
* 主动确认单笔订单支付状态。
*
* @param outTradeNumber 商户订单号
* @return 是否已确认并处理支付成功
*/
boolean confirmPaidOrder(String outTradeNumber);
/**
* 修复已支付订单中设备有效期未落库的异常数据。
*
* @return 修复的设备数量
*/
int repairPaidOrderDeviceDeadTime();
}

View File

@@ -61,4 +61,13 @@ public interface WxPayService {
* @return 解密后的明文
*/
String decryptCallbackData(String associatedData, String nonce, String ciphertext);
/**
* 主动查询微信支付订单状态
* 通过商户订单号向微信支付平台查询订单的实际支付状态
*
* @param outTradeNumber 商户订单号
* @return 微信返回的订单JSON字符串查询失败返回null
*/
String queryOrderByOutTradeNo(String outTradeNumber);
}

View File

@@ -1,9 +1,10 @@
package com.intc.weixin.service.impl;
import cn.hutool.core.date.DateUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.intc.common.core.exception.ServiceException;
import com.intc.common.tenant.helper.TenantHelper;
import com.intc.fishery.constant.DefineDeviceWarnCode;
import com.intc.fishery.domain.Device;
import com.intc.fishery.domain.PayDevice;
@@ -14,16 +15,22 @@ import com.intc.fishery.mapper.PayOrderMapper;
import com.intc.weixin.config.WxPayItemProperties;
import com.intc.weixin.constant.PayOrderStatus;
import com.intc.weixin.service.PayOrderBusinessService;
import com.intc.weixin.service.WxPayService;
import com.intc.weixin.utils.OrderNumberGenerator;
import com.intc.weixin.utils.PayOrderStatusUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionTemplate;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
/**
* 支付订单业务服务实现
@@ -39,11 +46,20 @@ public class PayOrderBusinessServiceImpl implements PayOrderBusinessService {
private final PayDeviceMapper payDeviceMapper;
private final DeviceMapper deviceMapper;
private final WxPayItemProperties wxPayItemProperties;
private final WxPayService wxPayService;
private final TransactionTemplate transactionTemplate;
/**
* 订单处理锁,防止并发处理同一订单
*/
private static final ConcurrentHashMap<Long, Long> ORDER_PROCESSING_LOCK = new ConcurrentHashMap<>();
private static final AtomicLong COMPENSATION_HISTORY_CURSOR_ID = new AtomicLong(0L);
private static final AtomicLong REPAIR_HISTORY_CURSOR_ID = new AtomicLong(0L);
private static final int RECENT_SCAN_SIZE = 200;
private static final int HISTORY_SCAN_SIZE = 200;
private static final int ORDER_UPDATE_FAILED = -1;
private static final int ORDER_ALREADY_SUCCESS = 0;
private static final int ORDER_UPDATED = 1;
@Override
@Transactional(rollbackFor = Exception.class)
@@ -54,14 +70,50 @@ public class PayOrderBusinessServiceImpl implements PayOrderBusinessService {
throw new ServiceException("支付选项不存在");
}
// 2. 验证设备归属
List<Device> devices = deviceMapper.selectList(
new LambdaQueryWrapper<Device>()
.in(Device::getId, deviceIds)
.eq(Device::getUserId, userId)
// 2. 去重设备ID防止前端传入重复ID导致 size 不匹配)
List<Long> uniqueDeviceIds = deviceIds.stream().distinct().collect(Collectors.toList());
if (uniqueDeviceIds.size() != deviceIds.size()) {
log.warn("设备ID列表存在重复原始数量={},去重后数量={}", deviceIds.size(), uniqueDeviceIds.size());
}
// 3. 验证设备归属忽略租户过滤因为设备可能由IOT模块写入租户ID可能不一致
List<Device> devices = TenantHelper.ignore(() ->
deviceMapper.selectList(
new LambdaQueryWrapper<Device>()
.in(Device::getId, uniqueDeviceIds)
// .eq(Device::getUserId, userId)
)
);
if (devices == null || devices.size() != deviceIds.size()) {
if (devices == null || devices.size() != uniqueDeviceIds.size()) {
// 诊断日志:定位具体缺失的设备
Set<Long> foundIds = devices != null
? devices.stream().map(Device::getId).collect(Collectors.toSet())
: Collections.emptySet();
List<Long> missingIds = uniqueDeviceIds.stream()
.filter(id -> !foundIds.contains(id))
.collect(Collectors.toList());
log.error("设备归属验证失败: userId={}, 请求设备数={}, 找到设备数={}, 缺失设备ID={}",
userId, uniqueDeviceIds.size(), devices != null ? devices.size() : 0, missingIds);
// 进一步诊断检查缺失设备是否存在于数据库不限制userId
if (!missingIds.isEmpty()) {
List<Device> existDevices = TenantHelper.ignore(() ->
deviceMapper.selectList(
new LambdaQueryWrapper<Device>()
.in(Device::getId, missingIds)
)
);
if (existDevices != null && !existDevices.isEmpty()) {
for (Device d : existDevices) {
log.error("设备存在但不属于当前用户: deviceId={}, deviceUserId={}, currentUserId={}, tenantId={}",
d.getId(), d.getUserId(), userId, d.getTenantId());
}
} else {
log.error("设备在数据库中不存在: missingIds={}", missingIds);
}
}
throw new ServiceException("部分设备不存在或无权限操作");
}
@@ -134,7 +186,6 @@ public class PayOrderBusinessServiceImpl implements PayOrderBusinessService {
}
@Override
@Transactional(rollbackFor = Exception.class)
public void closeUnpaidOrders(Long userId, String openId) {
// 查询用户所有未完成的订单
List<PayOrder> unpaidOrders = payOrderMapper.selectList(
@@ -152,8 +203,50 @@ public class PayOrderBusinessServiceImpl implements PayOrderBusinessService {
return;
}
// 批量关闭订单
// 批量处理订单:先查微信确认状态,避免错误关闭已支付订单
for (PayOrder order : unpaidOrders) {
try {
String queryResult = wxPayService.queryOrderByOutTradeNo(order.getOutTradeNumber());
if (queryResult != null) {
JSONObject resultJson = JSONUtil.parseObj(queryResult);
String tradeState = resultJson.getStr("trade_state");
if ("SUCCESS".equalsIgnoreCase(tradeState)) {
// 微信侧已支付成功,执行补偿而不是关闭
log.info("关闭订单前查询发现已支付: orderId={}, outTradeNumber={}", order.getId(), order.getOutTradeNumber());
compensateSingleOrder(order, resultJson);
continue;
}
if ("CLOSED".equalsIgnoreCase(tradeState)) {
payOrderMapper.update(null,
new LambdaUpdateWrapper<PayOrder>()
.eq(PayOrder::getId, order.getId())
.set(PayOrder::getOrderStatus, PayOrderStatus.CLOSED)
.set(PayOrder::getTradeState, "CLOSED")
.set(PayOrder::getTradeStateDescription, "微信侧已关闭")
);
log.info("旧订单微信侧已关闭,同步本地状态: orderId={}, outTradeNumber={}",
order.getId(), order.getOutTradeNumber());
continue;
}
if ("USERPAYING".equalsIgnoreCase(tradeState)) {
log.info("旧订单仍在支付中,跳过关闭: orderId={}, outTradeNumber={}",
order.getId(), order.getOutTradeNumber());
continue;
}
}
} catch (Exception e) {
log.warn("关闭订单前查询微信失败,跳过关闭避免误关已支付订单: orderId={}, error={}", order.getId(), e.getMessage());
continue;
}
boolean closed = wxPayService.closeOrder(order.getOutTradeNumber());
if (!closed) {
log.warn("微信关单失败,跳过本地关闭: orderId={}, outTradeNumber={}",
order.getId(), order.getOutTradeNumber());
continue;
}
// 微信侧确认未支付,安全关闭
payOrderMapper.update(null,
new LambdaUpdateWrapper<PayOrder>()
.eq(PayOrder::getId, order.getId())
@@ -167,7 +260,6 @@ public class PayOrderBusinessServiceImpl implements PayOrderBusinessService {
}
@Override
@Transactional(rollbackFor = Exception.class)
public boolean handlePaymentSuccess(String outTradeNumber, String transactionId, Integer payerTotal,
String payerCurrency, String successTime, String tradeState,
String tradeStateDescription, String bankType) {
@@ -180,20 +272,20 @@ public class PayOrderBusinessServiceImpl implements PayOrderBusinessService {
// 2. 检查订单状态,如果已经是成功状态,直接返回
if (PayOrderStatus.SUCCESS.equals(order.getOrderStatus())) {
log.info("订单已处理过: orderId={}, outTradeNumber={}", order.getId(), outTradeNumber);
int repairCount = repairOrderDeviceDeadTime(order);
log.info("订单已处理过: orderId={}, outTradeNumber={}, repairDeviceCount={}",
order.getId(), outTradeNumber, repairCount);
return true;
}
// 3. 并发控制:检查订单是否正在处理
if (ORDER_PROCESSING_LOCK.containsKey(order.getId())) {
log.warn("订单正在处理中,跳过: orderId={}, outTradeNumber={}", order.getId(), outTradeNumber);
return false;
Long previousLockTime = ORDER_PROCESSING_LOCK.putIfAbsent(order.getId(), System.currentTimeMillis());
if (previousLockTime != null) {
log.warn("订单正在处理中,等待微信下次通知或补偿任务处理: orderId={}, outTradeNumber={}", order.getId(), outTradeNumber);
return true;
}
try {
// 4. 加锁
ORDER_PROCESSING_LOCK.put(order.getId(), System.currentTimeMillis());
// 5. 检查金额是否一致
if (!order.getTotalAmount().equals(payerTotal)) {
log.error("订单金额不匹配: orderId={}, orderAmount={}, payAmount={}",
@@ -206,22 +298,55 @@ public class PayOrderBusinessServiceImpl implements PayOrderBusinessService {
// 7. 更新订单状态
Integer orderStatus = PayOrderStatusUtil.convertTradeStateToOrderStatus(tradeState);
payOrderMapper.update(null,
new LambdaUpdateWrapper<PayOrder>()
.eq(PayOrder::getId, order.getId())
.set(PayOrder::getTransactionId, transactionId)
.set(PayOrder::getPayerTotal, payerTotal)
.set(PayOrder::getPayerCurrency, payerCurrency)
.set(PayOrder::getSuccessTime, successDate)
.set(PayOrder::getTradeState, tradeState)
.set(PayOrder::getTradeStateDescription, tradeStateDescription)
.set(PayOrder::getBankType, bankType)
.set(PayOrder::getOrderStatus, orderStatus)
);
Integer updateResult = transactionTemplate.execute(status -> {
int updateRows = payOrderMapper.update(null,
new LambdaUpdateWrapper<PayOrder>()
.eq(PayOrder::getId, order.getId())
.ne(PayOrder::getOrderStatus, PayOrderStatus.SUCCESS)
.set(PayOrder::getTransactionId, transactionId)
.set(PayOrder::getPayerTotal, payerTotal)
.set(PayOrder::getPayerCurrency, payerCurrency)
.set(PayOrder::getSuccessTime, successDate)
.set(PayOrder::getTradeState, tradeState)
.set(PayOrder::getTradeStateDescription, tradeStateDescription)
.set(PayOrder::getBankType, bankType)
.set(PayOrder::getOrderStatus, orderStatus)
);
if (updateRows <= 0) {
PayOrder latestOrder = payOrderMapper.selectById(order.getId());
if (latestOrder != null && PayOrderStatus.SUCCESS.equals(latestOrder.getOrderStatus())) {
log.info("订单已由其他回调处理成功: orderId={}, outTradeNumber={}", order.getId(), outTradeNumber);
return ORDER_ALREADY_SUCCESS;
}
log.warn("订单状态更新失败: orderId={}, outTradeNumber={}", order.getId(), outTradeNumber);
return ORDER_UPDATE_FAILED;
}
return ORDER_UPDATED;
});
if (updateResult == null || ORDER_UPDATE_FAILED == updateResult) {
return false;
}
// 8. 如果支付成功,更新设备到期时间
if (PayOrderStatus.SUCCESS.equals(orderStatus)) {
updateDeviceDeadTime(order.getId());
if (ORDER_UPDATED == updateResult) {
try {
int updatedDevices = updateDeviceDeadTime(order.getId());
if (updatedDevices <= 0) {
log.warn("订单已支付成功,但本次未更新任何设备有效期,等待修复任务兜底: orderId={}, outTradeNumber={}",
order.getId(), outTradeNumber);
}
} catch (Exception e) {
log.error("订单已支付成功,但更新设备有效期异常,等待修复任务兜底: orderId={}, outTradeNumber={}, error={}",
order.getId(), outTradeNumber, e.getMessage(), e);
}
} else {
int repairCount = repairOrderDeviceDeadTime(order);
log.info("订单已由其他实例处理成功,本次只做设备有效期修复: orderId={}, outTradeNumber={}, repairDeviceCount={}",
order.getId(), outTradeNumber, repairCount);
}
}
log.info("订单支付处理完成: orderId={}, outTradeNumber={}, status={}",
@@ -290,47 +415,101 @@ public class PayOrderBusinessServiceImpl implements PayOrderBusinessService {
*
* @param orderId 订单ID
*/
private void updateDeviceDeadTime(Long orderId) {
private int updateDeviceDeadTime(Long orderId) {
// 查询订单关联的设备充值记录
List<PayDevice> payDevices = queryPayDevicesByOrderId(orderId);
if (payDevices == null || payDevices.isEmpty()) {
log.warn("订单没有关联的设备充值记录: orderId={}", orderId);
return;
return 0;
}
// 批量更新设备到期时间
int updatedCount = 0;
// 批量更新设备到期时间忽略租户过滤因为设备可能由IOT模块写入租户ID可能不一致
for (PayDevice payDevice : payDevices) {
Device device = deviceMapper.selectOne(
new LambdaQueryWrapper<Device>()
.eq(Device::getSerialNum, payDevice.getSerialNum())
.eq(Device::getUserId, payDevice.getUserId())
);
try {
if (payDevice.getAddMonth() == null) {
log.error("设备充值记录缺少续费月数,跳过设备有效期更新: payDeviceId={}, orderId={}, serialNum={}",
payDevice.getId(), orderId, payDevice.getSerialNum());
continue;
}
if (device == null) {
log.warn("设备不存在: serialNum={}, userId={}", payDevice.getSerialNum(), payDevice.getUserId());
continue;
Device device = queryDeviceByPayDevice(payDevice);
if (device == null) {
log.error("设备不存在,订单成功状态已保留,等待人工核对设备编号: orderId={}, payDeviceId={}, serialNum={}",
orderId, payDevice.getId(), payDevice.getSerialNum());
continue;
}
// 计算新的到期时间
Date newDeadTime = calculateNewDeadTime(device.getDeadTime(), payDevice.getAddMonth());
payDeviceMapper.update(null,
new LambdaUpdateWrapper<PayDevice>()
.eq(PayDevice::getId, payDevice.getId())
.set(PayDevice::getDeadTime, newDeadTime)
);
// 清除 warnCode 中的「设备到期」标志位(续费后恢复正常状态)
Integer oldWarnCode = device.getWarnCode();
Integer newWarnCode = oldWarnCode != null
? (oldWarnCode & ~DefineDeviceWarnCode.DeviceTimeDead)
: 0;
// 更新设备到期时间,并同步清除到期告警位
int updateRows = TenantHelper.ignore(() -> deviceMapper.update(null,
new LambdaUpdateWrapper<Device>()
.eq(Device::getId, device.getId())
.set(Device::getDeadTime, newDeadTime)
.set(Device::getWarnCode, newWarnCode)
));
if (updateRows <= 0) {
log.error("更新设备有效期失败,订单成功状态已保留,等待修复任务兜底: orderId={}, payDeviceId={}, serialNum={}, deviceId={}",
orderId, payDevice.getId(), payDevice.getSerialNum(), device.getId());
continue;
}
updatedCount++;
log.info("续费成功,更新设备到期时间并清除到期告警: deviceId={}, serialNum={}, newDeadTime={}, oldWarnCode={}, newWarnCode={}",
device.getId(), device.getSerialNum(), newDeadTime, oldWarnCode, newWarnCode);
} catch (Exception e) {
log.error("更新单台设备有效期异常,订单成功状态已保留: orderId={}, payDeviceId={}, serialNum={}, error={}",
orderId, payDevice.getId(), payDevice.getSerialNum(), e.getMessage(), e);
}
}
return updatedCount;
}
private Device queryDeviceByPayDevice(PayDevice payDevice) {
return TenantHelper.ignore(() -> {
if (payDevice.getTenantId() != null && !payDevice.getTenantId().isEmpty()) {
List<Device> sameTenantDevices = deviceMapper.selectList(
new LambdaQueryWrapper<Device>()
.eq(Device::getSerialNum, payDevice.getSerialNum())
.eq(Device::getTenantId, payDevice.getTenantId())
.orderByDesc(Device::getUpdateTime)
.orderByDesc(Device::getCreateTime)
);
if (sameTenantDevices != null && !sameTenantDevices.isEmpty()) {
return sameTenantDevices.get(0);
}
}
// 计算新的到期时间
Date newDeadTime = calculateNewDeadTime(device.getDeadTime(), payDevice.getAddMonth());
// 清除 warnCode 中的「设备到期」标志位(续费后恢复正常状态)
Integer oldWarnCode = device.getWarnCode();
Integer newWarnCode = oldWarnCode != null
? (oldWarnCode & ~DefineDeviceWarnCode.DeviceTimeDead)
: 0;
// 更新设备到期时间,并同步清除到期告警位
deviceMapper.update(null,
new LambdaUpdateWrapper<Device>()
.eq(Device::getId, device.getId())
.set(Device::getDeadTime, newDeadTime)
.set(Device::getWarnCode, newWarnCode)
List<Device> devices = deviceMapper.selectList(
new LambdaQueryWrapper<Device>()
.eq(Device::getSerialNum, payDevice.getSerialNum())
.orderByDesc(Device::getUpdateTime)
.orderByDesc(Device::getCreateTime)
);
log.info("续费成功,更新设备到期时间并清除到期告警: deviceId={}, serialNum={}, newDeadTime={}, oldWarnCode={}, newWarnCode={}",
device.getId(), device.getSerialNum(), newDeadTime, oldWarnCode, newWarnCode);
}
if (devices == null || devices.isEmpty()) {
return null;
}
if (devices.size() > 1) {
log.warn("根据设备编号查到多台设备,已取最近更新的一台: serialNum={}, selectedDeviceId={}, matchedCount={}",
payDevice.getSerialNum(), devices.get(0).getId(), devices.size());
}
return devices.get(0);
});
}
/**
@@ -354,4 +533,338 @@ public class PayOrderBusinessServiceImpl implements PayOrderBusinessService {
return new Date();
}
}
@Override
public int compensateUnpaidOrders() {
// 最新订单优先处理,历史订单用游标慢慢补扫,避免老异常数据被固定时间窗口或分页上限漏掉。
Date now = new Date();
Date endTime = new Date(now.getTime() - 1L * 60 * 1000); // 1分钟前
Map<Long, PayOrder> candidateMap = new LinkedHashMap<>();
scanRecentCompensationOrders(endTime).forEach(order -> candidateMap.put(order.getId(), order));
scanHistoryCompensationOrders(endTime).forEach(order -> candidateMap.putIfAbsent(order.getId(), order));
if (candidateMap.isEmpty()) {
return 0;
}
int compensatedCount = 0;
for (PayOrder order : candidateMap.values()) {
if (compensateOrderFromWechat(order)) {
compensatedCount++;
}
}
log.info("[支付补偿] 本次扫描 {} 笔待补偿订单", candidateMap.size());
if (compensatedCount > 0) {
log.info("[支付补偿] 本次补偿完成 {} 笔订单", compensatedCount);
}
return compensatedCount;
}
private List<PayOrder> scanRecentCompensationOrders(Date endTime) {
Page<PayOrder> page = payOrderMapper.selectPage(
new Page<>(1, RECENT_SCAN_SIZE, false),
buildCompensationOrderQuery(endTime)
.orderByDesc(PayOrder::getCreateTime)
.orderByDesc(PayOrder::getId)
);
return page.getRecords() != null ? page.getRecords() : Collections.emptyList();
}
private List<PayOrder> scanHistoryCompensationOrders(Date endTime) {
long cursorId = COMPENSATION_HISTORY_CURSOR_ID.get();
LambdaQueryWrapper<PayOrder> queryWrapper = buildCompensationOrderQuery(endTime);
if (cursorId > 0) {
queryWrapper.gt(PayOrder::getId, cursorId);
}
Page<PayOrder> page = payOrderMapper.selectPage(
new Page<>(1, HISTORY_SCAN_SIZE, false),
queryWrapper.orderByAsc(PayOrder::getId)
);
List<PayOrder> records = page.getRecords();
if (records == null || records.isEmpty()) {
if (cursorId > 0) {
COMPENSATION_HISTORY_CURSOR_ID.set(0L);
log.info("[支付补偿] 历史扫描游标已到末尾,下次从最早订单重新扫描");
}
return Collections.emptyList();
}
COMPENSATION_HISTORY_CURSOR_ID.set(records.get(records.size() - 1).getId());
return records;
}
private LambdaQueryWrapper<PayOrder> buildCompensationOrderQuery(Date endTime) {
return new LambdaQueryWrapper<PayOrder>()
.and(wrapper -> wrapper
.in(PayOrder::getOrderStatus, Arrays.asList(
PayOrderStatus.NOTPAY,
PayOrderStatus.USERPAYING,
PayOrderStatus.PAYERROR
))
.or()
.and(closed -> closed
.eq(PayOrder::getOrderStatus, PayOrderStatus.CLOSED)
.and(unconfirmedClosed -> unconfirmedClosed
.isNull(PayOrder::getTradeStateDescription)
.or()
.ne(PayOrder::getTradeStateDescription, "微信侧已关闭")
)
)
.or()
.isNull(PayOrder::getOrderStatus)
)
.isNotNull(PayOrder::getOutTradeNumber)
.ne(PayOrder::getOutTradeNumber, "")
.and(time -> time
.isNull(PayOrder::getCreateTime)
.or()
.le(PayOrder::getCreateTime, endTime)
);
}
private boolean compensateOrderFromWechat(PayOrder order) {
try {
String queryResult = wxPayService.queryOrderByOutTradeNo(order.getOutTradeNumber());
if (queryResult == null) {
log.warn("[支付补偿] 查询微信订单失败: orderId={}, outTradeNumber={}",
order.getId(), order.getOutTradeNumber());
return false;
}
JSONObject resultJson = JSONUtil.parseObj(queryResult);
String tradeState = resultJson.getStr("trade_state");
log.info("[支付补偿] 微信查单结果: orderId={}, outTradeNumber={}, localStatus={}, tradeState={}, prepayId={}, createTime={}",
order.getId(), order.getOutTradeNumber(), order.getOrderStatus(), tradeState, order.getPrepayId(), order.getCreateTime());
if ("SUCCESS".equalsIgnoreCase(tradeState)) {
boolean success = compensateSingleOrder(order, resultJson);
if (success) {
log.info("[支付补偿] 补偿成功: orderId={}, outTradeNumber={}", order.getId(), order.getOutTradeNumber());
} else {
log.warn("[支付补偿] 微信已支付但本地补偿失败: orderId={}, outTradeNumber={}",
order.getId(), order.getOutTradeNumber());
}
return success;
}
if ("CLOSED".equalsIgnoreCase(tradeState)) {
syncLocalClosedOrder(order.getId(), "微信侧已关闭");
}
return false;
} catch (Exception e) {
log.error("[支付补偿] 处理订单异常: orderId={}, outTradeNumber={}, error={}",
order.getId(), order.getOutTradeNumber(), e.getMessage(), e);
return false;
}
}
private void syncLocalClosedOrder(Long orderId, String description) {
payOrderMapper.update(null,
new LambdaUpdateWrapper<PayOrder>()
.eq(PayOrder::getId, orderId)
.ne(PayOrder::getOrderStatus, PayOrderStatus.SUCCESS)
.set(PayOrder::getOrderStatus, PayOrderStatus.CLOSED)
.set(PayOrder::getTradeState, "CLOSED")
.set(PayOrder::getTradeStateDescription, description)
);
log.info("[支付补偿] 订单微信侧已关闭,同步关闭: orderId={}", orderId);
}
/**
* 对单笔订单执行支付补偿
*
* @param order 本地订单
* @param wxResult 微信查询结果
* @return 是否补偿成功
*/
private boolean compensateSingleOrder(PayOrder order, JSONObject wxResult) {
String transactionId = wxResult.getStr("transaction_id");
String tradeState = wxResult.getStr("trade_state");
String tradeStateDesc = wxResult.getStr("trade_state_desc");
String successTime = wxResult.getStr("success_time");
String bankType = wxResult.getStr("bank_type");
JSONObject amountObj = wxResult.getJSONObject("amount");
Integer payerTotal = amountObj != null ? amountObj.getInt("payer_total") : null;
String payerCurrency = amountObj != null ? amountObj.getStr("payer_currency") : "CNY";
if (payerTotal == null) {
log.error("[支付补偿] 微信返回金额为空: orderId={}", order.getId());
return false;
}
return handlePaymentSuccess(
order.getOutTradeNumber(),
transactionId,
payerTotal,
payerCurrency,
successTime,
tradeState,
tradeStateDesc,
bankType
);
}
@Override
public boolean confirmPaidOrder(String outTradeNumber) {
if (outTradeNumber == null || outTradeNumber.isEmpty()) {
return false;
}
PayOrder order = queryByOutTradeNumber(outTradeNumber);
if (order == null) {
log.warn("[支付确认] 本地订单不存在: outTradeNumber={}", outTradeNumber);
return false;
}
if (PayOrderStatus.SUCCESS.equals(order.getOrderStatus())) {
int repairCount = repairOrderDeviceDeadTime(order);
log.info("[支付确认] 订单已成功: orderId={}, outTradeNumber={}, repairDeviceCount={}",
order.getId(), outTradeNumber, repairCount);
return true;
}
String queryResult = wxPayService.queryOrderByOutTradeNo(outTradeNumber);
if (queryResult == null) {
log.warn("[支付确认] 查询微信订单失败: orderId={}, outTradeNumber={}", order.getId(), outTradeNumber);
return false;
}
JSONObject resultJson = JSONUtil.parseObj(queryResult);
String tradeState = resultJson.getStr("trade_state");
if ("SUCCESS".equalsIgnoreCase(tradeState)) {
return compensateSingleOrder(order, resultJson);
}
if ("CLOSED".equalsIgnoreCase(tradeState)) {
payOrderMapper.update(null,
new LambdaUpdateWrapper<PayOrder>()
.eq(PayOrder::getId, order.getId())
.ne(PayOrder::getOrderStatus, PayOrderStatus.SUCCESS)
.set(PayOrder::getOrderStatus, PayOrderStatus.CLOSED)
.set(PayOrder::getTradeState, "CLOSED")
.set(PayOrder::getTradeStateDescription, "微信侧已关闭")
);
}
log.info("[支付确认] 订单未支付成功: orderId={}, outTradeNumber={}, tradeState={}",
order.getId(), outTradeNumber, tradeState);
return false;
}
@Override
@Transactional(rollbackFor = Exception.class)
public int repairPaidOrderDeviceDeadTime() {
Map<Long, PayOrder> paidOrderMap = new LinkedHashMap<>();
scanRecentPaidOrdersForRepair().forEach(order -> paidOrderMap.put(order.getId(), order));
scanHistoryPaidOrdersForRepair().forEach(order -> paidOrderMap.putIfAbsent(order.getId(), order));
if (paidOrderMap.isEmpty()) {
return 0;
}
int repairCount = 0;
for (PayOrder order : paidOrderMap.values()) {
repairCount += repairOrderDeviceDeadTime(order);
}
if (repairCount > 0) {
log.warn("[支付有效期修复] 本次修复 {} 台设备", repairCount);
} else {
log.info("[支付有效期修复] 本次扫描 {} 笔支付成功订单,无需修复", paidOrderMap.size());
}
return repairCount;
}
private List<PayOrder> scanRecentPaidOrdersForRepair() {
Page<PayOrder> page = payOrderMapper.selectPage(
new Page<>(1, RECENT_SCAN_SIZE, false),
buildPaidOrderRepairQuery()
.orderByDesc(PayOrder::getCreateTime)
.orderByDesc(PayOrder::getId)
);
return page.getRecords() != null ? page.getRecords() : Collections.emptyList();
}
private List<PayOrder> scanHistoryPaidOrdersForRepair() {
long cursorId = REPAIR_HISTORY_CURSOR_ID.get();
LambdaQueryWrapper<PayOrder> queryWrapper = buildPaidOrderRepairQuery();
if (cursorId > 0) {
queryWrapper.gt(PayOrder::getId, cursorId);
}
Page<PayOrder> page = payOrderMapper.selectPage(
new Page<>(1, HISTORY_SCAN_SIZE, false),
queryWrapper.orderByAsc(PayOrder::getId)
);
List<PayOrder> records = page.getRecords();
if (records == null || records.isEmpty()) {
if (cursorId > 0) {
REPAIR_HISTORY_CURSOR_ID.set(0L);
log.info("[支付有效期修复] 历史扫描游标已到末尾,下次从最早成功订单重新扫描");
}
return Collections.emptyList();
}
REPAIR_HISTORY_CURSOR_ID.set(records.get(records.size() - 1).getId());
return records;
}
private LambdaQueryWrapper<PayOrder> buildPaidOrderRepairQuery() {
return new LambdaQueryWrapper<PayOrder>()
.eq(PayOrder::getOrderStatus, PayOrderStatus.SUCCESS);
}
private int repairOrderDeviceDeadTime(PayOrder order) {
List<PayDevice> payDevices = queryPayDevicesByOrderId(order.getId());
if (payDevices == null || payDevices.isEmpty()) {
log.warn("[支付有效期修复] 支付成功订单没有设备充值记录: orderId={}", order.getId());
return 0;
}
int repairCount = 0;
for (PayDevice payDevice : payDevices) {
try {
if (payDevice.getDeadTime() == null) {
log.warn("[支付有效期修复] 设备充值记录没有目标到期时间: payDeviceId={}, orderId={}",
payDevice.getId(), order.getId());
continue;
}
Device device = queryDeviceByPayDevice(payDevice);
if (device == null) {
log.warn("[支付有效期修复] 设备不存在: orderId={}, serialNum={}",
order.getId(), payDevice.getSerialNum());
continue;
}
if (device.getDeadTime() != null && !device.getDeadTime().before(payDevice.getDeadTime())) {
continue;
}
Integer oldWarnCode = device.getWarnCode();
Integer newWarnCode = oldWarnCode != null
? (oldWarnCode & ~DefineDeviceWarnCode.DeviceTimeDead)
: 0;
int updateRows = TenantHelper.ignore(() -> deviceMapper.update(null,
new LambdaUpdateWrapper<Device>()
.eq(Device::getId, device.getId())
.set(Device::getDeadTime, payDevice.getDeadTime())
.set(Device::getWarnCode, newWarnCode)
));
if (updateRows > 0) {
repairCount++;
log.warn("[支付有效期修复] 已修复设备有效期: orderId={}, deviceId={}, serialNum={}, oldDeadTime={}, targetDeadTime={}",
order.getId(), device.getId(), device.getSerialNum(), device.getDeadTime(), payDevice.getDeadTime());
}
} catch (Exception e) {
log.error("[支付有效期修复] 修复单台设备异常: orderId={}, payDeviceId={}, serialNum={}, error={}",
order.getId(), payDevice.getId(), payDevice.getSerialNum(), e.getMessage(), e);
}
}
return repairCount;
}
}

View File

@@ -36,6 +36,7 @@ public class WxPayServiceImpl implements WxPayService {
private static final String JSAPI_URL = "https://api.mch.weixin.qq.com/v3/pay/transactions/jsapi";
private static final String CLOSE_URL_TEMPLATE = "https://api.mch.weixin.qq.com/v3/pay/transactions/out-trade-no/%s/close";
private static final String QUERY_ORDER_URL_TEMPLATE = "https://api.mch.weixin.qq.com/v3/pay/transactions/out-trade-no/%s?mchid=%s";
private final WxPayProperties wxPayProperties;
private final WxMaProperties wxMaProperties;
@@ -126,6 +127,31 @@ public class WxPayServiceImpl implements WxPayService {
return response.body();
}
// ----------------------------------------------------------------
// 发送 HTTP GET
// ----------------------------------------------------------------
private String doGet(String url) throws Exception {
URI uri = URI.create(url);
String urlPath = uri.getPath();
if (uri.getQuery() != null) {
urlPath = urlPath + "?" + uri.getQuery();
}
String authorization = buildAuthorization("GET", urlPath, "");
HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
.uri(uri)
.header("Accept", "application/json")
.header("Authorization", authorization)
.GET()
.build();
HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
log.info("微信支付查询响应: status={}, body={}", response.statusCode(), response.body());
if (response.statusCode() < 200 || response.statusCode() >= 300) {
throw new RuntimeException("微信支付查询失败: " + response.statusCode() + " " + response.body());
}
return response.body();
}
// ----------------------------------------------------------------
// 发送 HTTP POST无响应体如关闭订单
// ----------------------------------------------------------------
@@ -155,7 +181,7 @@ public class WxPayServiceImpl implements WxPayService {
@Override
public String createJsapiOrder(String openId, String outTradeNumber, Integer totalAmount,
String description, String notifyUrl) {
log.info("创建JSAPI支付订单: openId={}, outTradeNumber={}, amount={}", openId, outTradeNumber, totalAmount);
log.info("创建JSAPI支付订单: openId={}, outTradeNumber={}, amount={}, notifyUrl={}", openId, outTradeNumber, totalAmount, notifyUrl);
try {
String appId = wxMaProperties.getAppId();
String mchId = wxPayProperties.getMchId();
@@ -283,4 +309,16 @@ public class WxPayServiceImpl implements WxPayService {
if (value == null) return "";
return value.replace("\\", "\\\\").replace("\"", "\\\"");
}
@Override
public String queryOrderByOutTradeNo(String outTradeNumber) {
log.info("主动查询订单状态: outTradeNumber={}", outTradeNumber);
try {
String url = String.format(QUERY_ORDER_URL_TEMPLATE, outTradeNumber, wxPayProperties.getMchId());
return doGet(url);
} catch (Exception e) {
log.error("查询订单状态失败: outTradeNumber={}, error={}", outTradeNumber, e.getMessage());
return null;
}
}
}

View File

@@ -0,0 +1,59 @@
package com.intc.weixin.task;
import com.intc.weixin.service.PayOrderBusinessService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
/**
* 支付订单补偿定时任务
* 定期扫描状态为"未支付/支付中/支付失败/已关闭"且有商户订单号的订单,
* 主动向微信支付查询实际状态,对已支付成功但回调丢失的订单进行补偿处理。
*
* 解决问题:用户实际已付款成功,但因网络/服务器等原因未收到微信支付回调通知,
* 导致订单一直停留在未支付状态。
*
* @author intc
*/
@Slf4j
@Component
@RequiredArgsConstructor
@ConditionalOnBean(PayOrderBusinessService.class)
public class PayOrderCompensationTask {
private final PayOrderBusinessService payOrderBusinessService;
/**
* 每1分钟执行一次订单补偿扫描
* 优先扫描最近订单,同时通过历史游标持续补扫老异常订单。
*/
// @Scheduled(fixedDelay = 60000, initialDelay = 60000)
public void compensateOrders() {
try {
int count = payOrderBusinessService.compensateUnpaidOrders();
if (count > 0) {
log.info("[支付补偿定时任务] 本次补偿了 {} 笔订单", count);
}
} catch (Exception e) {
log.error("[支付补偿定时任务] 执行异常: {}", e.getMessage(), e);
}
}
/**
* 每10分钟修复一次已支付订单的设备有效期。
* 优先扫描最近成功订单,同时通过历史游标持续补扫老成功订单。
*/
// @Scheduled(fixedDelay = 600000, initialDelay = 120000)
public void repairDeviceDeadTime() {
try {
int count = payOrderBusinessService.repairPaidOrderDeviceDeadTime();
if (count > 0) {
log.warn("[支付有效期修复任务] 本次修复了 {} 台设备", count);
}
} catch (Exception e) {
log.error("[支付有效期修复任务] 执行异常: {}", e.getMessage(), e);
}
}
}