Microservice/govern
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master'

Browse Source 
# Conflicts:
#	cs-device/cs-device-api/src/main/java/com/njcn/csdevice/pojo/po/CsEdDataPO.java
#	cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/CsEquipmentDeliveryServiceImpl.java
This commit is contained in:
caozehui
2026-04-22 08:52:53 +08:00
parent 00ccff18c0 8841000989
commit ae220ceeea
46 changed files with 1554 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
cs-device/cs-device-api/src/main/java/com/njcn/csdevice/api/DeviceMessageFeignClient.java
View File

@@ -27,4 +27,8 @@ public interface DeviceMessageFeignClient {
@ApiOperation("根据事件类型和用户id查询打开推送的用户信息")
HttpResult<List<User>> getSendUserByType(@RequestBody DeviceMessageParam param);
@PostMapping("/getLineInfo")
@ApiOperation("获取监测点信息")
HttpResult<String> getLineInfo(@RequestParam("id") String id);
}

23
cs-device/cs-device-api/src/main/java/com/njcn/csdevice/api/SmsSendFeignClient.java Normal file
View File

@@ -0,0 +1,23 @@
package com.njcn.csdevice.api;
import com.njcn.common.pojo.constant.ServerInfo;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.csdevice.api.fallback.SmsSendClientFallbackFactory;
import io.swagger.annotations.ApiOperation;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
/**
* @author xy
*/
@FeignClient(value = ServerInfo.CS_DEVICE_BOOT, path = "/sms", fallbackFactory = SmsSendClientFallbackFactory.class,contextId = "sms")
public interface SmsSendFeignClient {
@PostMapping("/send/simple")
@ApiOperation("发送短信（简化参数）")
HttpResult<String> sendSmsSimple(@RequestParam("receiver") String receiver
, @RequestParam("content") String content
, @RequestParam("messageType") String messageType);
}

6
cs-device/cs-device-api/src/main/java/com/njcn/csdevice/api/fallback/DeviceMessageClientFallbackFactory.java
View File

@@ -39,6 +39,12 @@ public class DeviceMessageClientFallbackFactory implements FallbackFactory<Devic
log.error("{}异常，降级处理，异常为：{}","根据事件类型和用户id查询打开推送的用户信息数据异常",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
@Override
public HttpResult<String> getLineInfo(String id) {
log.error("{}异常，降级处理，异常为：{}","获取监测点信息数据异常",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

34
cs-device/cs-device-api/src/main/java/com/njcn/csdevice/api/fallback/SmsSendClientFallbackFactory.java Normal file
View File

@@ -0,0 +1,34 @@
package com.njcn.csdevice.api.fallback;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.csdevice.api.SmsSendFeignClient;
import feign.hystrix.FallbackFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
/**
* @author xy
*/
@Slf4j
@Component
public class SmsSendClientFallbackFactory implements FallbackFactory<SmsSendFeignClient> {
@Override
public SmsSendFeignClient create(Throwable cause) {
//判断抛出异常是否为解码器抛出的业务异常
Enum<?> exceptionEnum = CommonResponseEnum.SERVICE_FALLBACK;
if (cause.getCause() instanceof BusinessException) {
BusinessException businessException = (BusinessException) cause.getCause();
}
Enum<?> finalExceptionEnum = exceptionEnum;
return new SmsSendFeignClient() {
@Override
public HttpResult<String> sendSmsSimple(String receiver, String content, String messageType) {
log.error("{}异常，降级处理，异常为：{}","发送短信（简化参数）数据异常",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

35
cs-device/cs-device-api/src/main/java/com/njcn/csdevice/pojo/dto/CredentialReqDTO.java Normal file
View File

@@ -0,0 +1,35 @@
package com.njcn.csdevice.pojo.dto;
/**
* @author caozehui
* @data 2026-03-31
*/
import lombok.Data;
import javax.validation.constraints.NotEmpty;
import java.io.Serializable;
/**
* 系统凭证请求 DTO
*
* @author msgpush
*/
@Data
public class CredentialReqDTO implements Serializable {
private static final long serialVersionUID = 1L;
/**
* 上游系统名称
*/
@NotEmpty(message = "上游系统名称不能为空")
private String systemName;
/**
* 密钥（用于生成凭证）
*/
@NotEmpty(message = "密钥不能为空")
private String secretKey;
}

21
cs-device/cs-device-api/src/main/java/com/njcn/csdevice/pojo/dto/SendResult.java Normal file
View File

@@ -0,0 +1,21 @@
package com.njcn.csdevice.pojo.dto;
import lombok.AllArgsConstructor;
import lombok.Data;
import java.io.Serializable;
/**
* @author xy
*/
@Data
@AllArgsConstructor
public class SendResult implements Serializable {
private final boolean success;
private final String messageId;
private final String failReason;
private final boolean isTimeOut;
private final boolean unauthorized;
}

2
cs-device/cs-device-api/src/main/java/com/njcn/csdevice/pojo/param/CsEdDataAuditParm.java
View File

@@ -67,7 +67,7 @@ public class CsEdDataAuditParm {
private String versionType;
@ApiModelProperty(value = "crc信息")
private String crcInfo;
private String crc;
@ApiModelProperty(value="0：删除 1：正常")
private String status;
@ApiModelProperty(value = ".bin文件")

4
cs-device/cs-device-api/src/main/java/com/njcn/csdevice/pojo/po/CsEdDataPO.java
View File

@@ -88,9 +88,11 @@ public class CsEdDataPO extends BaseEntity {
private String filePath;
/**
* crc校验
* crc文件校验码
*/
@TableField(value = "crc")
private String crc;
}

45
cs-device/cs-device-api/src/main/java/com/njcn/csdevice/pojo/po/CsSmsSendRecord.java Normal file
View File

@@ -0,0 +1,45 @@
package com.njcn.csdevice.pojo.po;
import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;
import java.io.Serializable;
import java.time.LocalDateTime;
/**
* @author xy
*/
@Data
@TableName("cs_sms_send_record")
public class CsSmsSendRecord implements Serializable {
private static final long serialVersionUID = 1L;
@TableId(type = IdType.ASSIGN_ID)
private String id;
private String receiver;
private String content;
private String messageType;
private String credentialToken;
private Integer sendStatus;
@TableField(updateStrategy = FieldStrategy.IGNORED)
private String failReason;
private Integer retryCount;
private Integer maxRetry;
private Long responseTime;
private LocalDateTime sendTime;
private LocalDateTime createTime;
private LocalDateTime updateTime;
}

40
cs-device/cs-device-api/src/main/java/com/njcn/csdevice/pojo/vo/MessageRecordReqVO.java Normal file
View File

@@ -0,0 +1,40 @@
package com.njcn.csdevice.pojo.vo;
import cn.hutool.core.lang.RegexPool;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
import javax.validation.constraints.NotBlank;
import javax.validation.constraints.Pattern;
/**
* @author caozehui
* @data 2026-02-27
*/
@Data
@Schema(description = "管理后台 - 消息记录发送 Request VO")
public class MessageRecordReqVO {
private String channel;
@Schema(description = "消息类型", example = "verify_code/order_notify/marketing/system_notify")
@NotBlank(message = "消息类型不能为空")
private String messageType;
@Schema(description = "接收者")
@NotBlank(message = "接收者不能为空")
@Pattern(regexp = RegexPool.EMAIL + "|" + RegexPool.MOBILE, message = "必须是有效的邮箱或手机号格式")
private String receiver;
@Schema(description = "标题")
private String title;
@Schema(description = "消息内容")
private String content;
@Schema(description = "模板编码")
private String templateCode;
@Schema(description = "模板参数")
private String templateParams;
}

148
cs-device/cs-device-api/src/main/java/com/njcn/csdevice/utils/MqttTest.java
View File

@@ -1,74 +1,74 @@
package com.njcn.csdevice.utils;
import org.eclipse.paho.client.mqttv3.*;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
/**
* Description:
* Date: 2023/8/2 13:41【需求编号】
*
* @author clam
* @version V1.0.0
*/
public class MqttTest {
private static final String MQTT_BROKER = "tcp://192.168.1.13:1883";
private static final String MQTT_TOPIC = "file/upload";
private static final String FILE_PATH = "C:\\Users\\无名\\Desktop\\111.json"; // Replace with the path to your file
public static void main(String[] args) {
MqttClient mqttClient = null;
try {
// Connect to the MQTT broker
mqttClient = new MqttClient(MQTT_BROKER, MqttClient.generateClientId());
MqttConnectOptions connOpts = new MqttConnectOptions();
connOpts.setUserName("t_user");
connOpts.setPassword("njcnpqs".toCharArray());
mqttClient.connect(connOpts);
// Read the file
File file = new File(FILE_PATH);
FileInputStream fis = new FileInputStream(file);
byte[] fileContent = new byte[(int) file.length()];
fis.read(fileContent);
fis.close();
// Create a new MQTT message
MqttMessage message = new MqttMessage(fileContent);
// Set QoS level and retain flag as per your requirement
message.setQos(1);
// message.setRetained(false);
// Record the start time
long startTime = System.currentTimeMillis();
// Publish the message to the MQTT topic
mqttClient.publish(MQTT_TOPIC, message);
// Record the end time
long endTime = System.currentTimeMillis();
System.out.println("File published successfully!");
System.out.println("Time taken: " + (endTime - startTime) + " ms");
} catch (MqttException | IOException e) {
e.printStackTrace();
} finally {
// Disconnect from the MQTT broker
if (mqttClient != null && mqttClient.isConnected()) {
try {
mqttClient.disconnect();
} catch (MqttException e) {
e.printStackTrace();
}
}
}
}
}
//package com.njcn.csdevice.utils;
//
//import org.eclipse.paho.client.mqttv3.*;
//
//import java.io.File;
//import java.io.FileInputStream;
//import java.io.FileOutputStream;
//import java.io.IOException;
//import java.nio.file.Files;
//import java.nio.file.Paths;
//
///**
// * Description:
// * Date: 2023/8/2 13:41【需求编号】
// *
// * @author clam
// * @version V1.0.0
// */
//public class MqttTest {
// private static final String MQTT_BROKER = "tcp://192.168.1.13:1883";
// private static final String MQTT_TOPIC = "file/upload";
// private static final String FILE_PATH = "C:\\Users\\无名\\Desktop\\111.json"; // Replace with the path to your file
//
// public static void main(String[] args) {
// MqttClient mqttClient = null;
// try {
// // Connect to the MQTT broker
// mqttClient = new MqttClient(MQTT_BROKER, MqttClient.generateClientId());
// MqttConnectOptions connOpts = new MqttConnectOptions();
// connOpts.setUserName("t_user");
// connOpts.setPassword("njcnpqs".toCharArray());
//
// mqttClient.connect(connOpts);
//
// // Read the file
// File file = new File(FILE_PATH);
// FileInputStream fis = new FileInputStream(file);
// byte[] fileContent = new byte[(int) file.length()];
// fis.read(fileContent);
// fis.close();
//
// // Create a new MQTT message
// MqttMessage message = new MqttMessage(fileContent);
//
// // Set QoS level and retain flag as per your requirement
// message.setQos(1);
//// message.setRetained(false);
//
// // Record the start time
// long startTime = System.currentTimeMillis();
//
// // Publish the message to the MQTT topic
// mqttClient.publish(MQTT_TOPIC, message);
//
// // Record the end time
// long endTime = System.currentTimeMillis();
//
// System.out.println("File published successfully!");
// System.out.println("Time taken: " + (endTime - startTime) + " ms");
// } catch (MqttException | IOException e) {
// e.printStackTrace();
// } finally {
// // Disconnect from the MQTT broker
// if (mqttClient != null && mqttClient.isConnected()) {
// try {
// mqttClient.disconnect();
// } catch (MqttException e) {
// e.printStackTrace();
// }
// }
// }
// }
//
//}

4
cs-device/cs-device-boot/pom.xml
View File

@@ -21,6 +21,10 @@
</properties>
<dependencies>
<dependency>
<groupId>com.github.tocrhz</groupId>
<artifactId>mqtt-spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>common-web</artifactId>

10
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/controller/message/DeviceMessageController.java
View File

@@ -66,5 +66,15 @@ public class DeviceMessageController extends BaseController {
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, list, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/getLineInfo")
@ApiOperation("获取监测点信息")
@ApiImplicitParam(name = "id", value = "参数", required = true, paramType = "query")
public HttpResult<String> getLineInfo(@RequestParam("id") String id){
String methodDescribe = getMethodDescribe("getLineInfo");
deviceMessageService.getLineInfo(id);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, "success", methodDescribe);
}
}

77
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/controller/message/SmsSendController.java Normal file
View File

@@ -0,0 +1,77 @@
package com.njcn.csdevice.controller.message;
import com.njcn.common.pojo.annotation.OperateInfo;
import com.njcn.common.pojo.enums.common.LogEnum;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.common.utils.HttpResultUtil;
import com.njcn.csdevice.pojo.vo.MessageRecordReqVO;
import com.njcn.csdevice.service.ISmsSendService;
import com.njcn.web.controller.BaseController;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*;
/**
* @author xy
*/
@Slf4j
@RestController
@RequestMapping("/sms")
@Api(tags = "短信发送管理")
@AllArgsConstructor
public class SmsSendController extends BaseController {
private final ISmsSendService smsSendService;
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/send")
@ApiOperation("发送短信（同步，包含重试）")
public HttpResult<String> sendSms(@RequestBody MessageRecordReqVO vo) {
String methodDescribe = getMethodDescribe("sendSms");
try {
smsSendService.sendSmsWithRetry(vo);
return HttpResultUtil.assembleCommonResponseResult(
CommonResponseEnum.SUCCESS,
"短信发送成功",
methodDescribe
);
} catch (Exception e) {
log.error("短信发送失败", e);
return HttpResultUtil.assembleCommonResponseResult(
CommonResponseEnum.FAIL,
e.getMessage(),
methodDescribe
);
}
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/send/simple")
@ApiOperation("发送短信（简化参数）")
public HttpResult<String> sendSmsSimple(
@RequestParam String receiver,
@RequestParam String content,
@RequestParam(defaultValue = "verify_code") String messageType) {
String methodDescribe = getMethodDescribe("sendSmsSimple");
try {
smsSendService.sendSmsWithRetry(receiver, content, messageType);
return HttpResultUtil.assembleCommonResponseResult(
CommonResponseEnum.SUCCESS,
"短信发送成功",
methodDescribe
);
} catch (Exception e) {
log.error("短信发送失败", e);
return HttpResultUtil.assembleCommonResponseResult(
CommonResponseEnum.FAIL,
e.getMessage(),
methodDescribe
);
}
}
}

7
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/mapper/CsSmsSendRecordMapper.java Normal file
View File

@@ -0,0 +1,7 @@
package com.njcn.csdevice.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.njcn.csdevice.pojo.po.CsSmsSendRecord;
public interface CsSmsSendRecordMapper extends BaseMapper<CsSmsSendRecord> {
}

2
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/DeviceMessageService.java
View File

@@ -14,4 +14,6 @@ public interface DeviceMessageService {
List<User> getSendUserByType(DeviceMessageParam param);
void getLineInfo(String id);
}

12
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/ISmsSendService.java Normal file
View File

@@ -0,0 +1,12 @@
package com.njcn.csdevice.service;
import com.baomidou.mybatisplus.extension.service.IService;
import com.njcn.csdevice.pojo.po.CsSmsSendRecord;
import com.njcn.csdevice.pojo.vo.MessageRecordReqVO;
public interface ISmsSendService extends IService<CsSmsSendRecord> {
void sendSmsWithRetry(String receiver, String content, String messageType);
void sendSmsWithRetry(MessageRecordReqVO messageRecordReqVO);
}

69
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/CsEquipmentDeliveryServiceImpl.java
View File

@@ -10,6 +10,7 @@ import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.text.StrPool;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONArray;
@@ -26,7 +27,6 @@ import com.njcn.access.utils.MqttUtil;
import com.njcn.common.pojo.dto.DeviceLogDTO;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.csdevice.api.CsLogsFeignClient;
import com.njcn.csdevice.api.EngineeringFeignClient;
import com.njcn.csdevice.constant.DataParam;
import com.njcn.csdevice.enums.AlgorithmResponseEnum;
import com.njcn.csdevice.mapper.CsEquipmentDeliveryMapper;
@@ -108,7 +108,6 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
private final AskDeviceDataFeignClient askDeviceDataFeignClient;
private final RedisUtil redisUtil;
private final CsSoftInfoMapper csSoftInfoMapper;
private final IMqttUserService mqttUserService;
private final MqttUtil mqttUtil;
private final CsLogsFeignClient csLogsFeignClient;
private final INodeService nodeService;
@@ -171,29 +170,40 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
csEquipmentProcess.setDevId(csEquipmentDeliveryAddParm.getNdid());
csEquipmentProcess.setOperator(RequestUtil.getUserIndex());
csEquipmentProcess.setStartTime(LocalDateTime.now());
// csEquipmentProcess.setProcess(1);
csEquipmentProcess.setProcess(4);
csEquipmentProcess.setStatus (1);
csEquipmentProcessPOService.save(csEquipmentProcess);
result = this.save (csEquipmentDeliveryPo);
//谁新建的设备，则认为是该设备的主用户，新建用户设备关系表数据
boolean addUser = csDeviceUserPOService.add(csEquipmentDeliveryPo.getId());
if (result && addUser) {
refreshDeviceDataCache();
String code = dictTreeFeignClient.queryById(csEquipmentDeliveryAddParm.getDevType()).getData().getCode();
if (Objects.equals(DicDataEnum.DEV_CLD.getCode(),code) && Objects.equals(csEquipmentDeliveryAddParm.getDevAccessMethod(),"CLD")) {
//谁新建的设备，则认为是该设备的主用户，新建用户设备关系表数据
boolean addUser = csDeviceUserPOService.add(csEquipmentDeliveryPo.getId());
if (result && addUser) {
refreshDeviceDataCache();
}
} else {
if (result) {
refreshDeviceDataCache();
}
}
return csEquipmentDeliveryPo;
}
@Override
@Transactional(rollbackFor = {Exception.class})
public Boolean AuditEquipmentDelivery(String id) {
CsEquipmentDeliveryPO one = this.lambdaQuery().eq(CsEquipmentDeliveryPO::getId, id).one();
//物理删除
QueryWrapper<CsEquipmentDeliveryPO> wrapper = new QueryWrapper();
wrapper.eq("id", id);
boolean update = this.remove(wrapper);
//删除deviceuser表里的设备，游客数据设备,删除监测点相关数据
boolean update = false;
LambdaQueryWrapper<CsEquipmentDeliveryPO> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(CsEquipmentDeliveryPO::getId,id);
CsEquipmentDeliveryPO po = this.getOne(wrapper);
if (po != null) {
update = this.remove(wrapper);
redisUtil.deleteKeysByString(AppRedisKey.LINE_POSITION+po.getNdid());
}
//删除监测点表、监测点拓扑图关系
List<CsLedger> list = csLedgerService.lambdaQuery().eq(CsLedger::getPid, id).list();
if (!CollectionUtils.isEmpty(list)) {
List<String> collect = list.stream().map(CsLedger::getId).collect(Collectors.toList());
@@ -204,23 +214,29 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
appLineTopologyDiagramPOQueryWrapper.clear();
appLineTopologyDiagramPOQueryWrapper.in("line_id", collect);
appLineTopologyDiagramService.remove(appLineTopologyDiagramPOQueryWrapper);
// appLineTopologyDiagramService.lambdaUpdate().in(AppLineTopologyDiagramPO::getLineId,collect).set(AppLineTopologyDiagramPO::getStatus,0).update();
}
LambdaQueryWrapper<CsLedger> csLedgerLambdaQueryWrapper = new LambdaQueryWrapper<>();
//清空关系表
LambdaQueryWrapper<CsLedger> csLedgerLambdaQueryWrapper = new LambdaQueryWrapper<>();
csLedgerLambdaQueryWrapper.clear();
csLedgerLambdaQueryWrapper.eq(CsLedger::getId, id);
csLedgerService.remove(csLedgerLambdaQueryWrapper);
csLedgerLambdaQueryWrapper.clear();
csLedgerLambdaQueryWrapper.eq(CsLedger::getPid, id);
csLedgerService.remove(csLedgerLambdaQueryWrapper);
csDeviceUserPOService.lambdaUpdate().eq(CsDeviceUserPO::getDeviceId, id).set(CsDeviceUserPO::getStatus, 0).update();
//删除设备和模板的关系
LambdaQueryWrapper<CsDevModelRelationPO> csDevModelRelationPOQueryWrapper = new LambdaQueryWrapper<>();
csDevModelRelationPOQueryWrapper.clear();
csDevModelRelationPOQueryWrapper.eq(CsDevModelRelationPO::getDevId,id);
csDevModelRelationService.remove(csDevModelRelationPOQueryWrapper);
//删除设备/用户关系
QueryWrapper<CsDeviceUserPO> csDeviceUserPOQueryWrapper = new QueryWrapper<>();
csDeviceUserPOQueryWrapper.clear();
csDeviceUserPOQueryWrapper.eq("device_id",id);
csDeviceUserPOService.remove(csDeviceUserPOQueryWrapper);
//删除游客
QueryWrapper<CsTouristDataPO> queryWrap = new QueryWrapper<>();
queryWrap.eq("device_id", id);
csTouristDataPOService.getBaseMapper().delete(queryWrap);
/*后续徐那边做处理*/
// CsEquipmentDeliveryPO csEquipmentDeliveryPO = this.getBaseMapper().selectById(id);
// mqttUserService.deleteUser(csEquipmentDeliveryPO.getNdid());
csDevModelRelationService.lambdaUpdate().eq(CsDevModelRelationPO::getDevId, id).set(CsDevModelRelationPO::getStatus, 0).update();
if (update) {
refreshDeviceDataCache();
}
@@ -389,11 +405,14 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
int total = this.baseMapper.getCounts(queryParam);
if (total > 0) {
List<CsEquipmentDeliveryVO> recordList = this.baseMapper.getList(queryParam);
//新增逻辑(只针对便携式设备):修改设备中的未注册状态(status = 1)改为5(前端定义的字典也即未接入)
for (CsEquipmentDeliveryVO csEquipmentDeliveryVO : recordList) {
if (DataParam.portableDevType.equals(csEquipmentDeliveryVO.getDevType()) && csEquipmentDeliveryVO.getStatus() == 1) {
//新增逻辑(针对便携式设备、监测设备):修改设备中的未注册状态(status = 1)改为5(前端定义的字典也即未接入)
for(CsEquipmentDeliveryVO csEquipmentDeliveryVO : recordList){
String code = dictTreeFeignClient.queryById(csEquipmentDeliveryVO.getDevType()).getData().getCode();
if((Objects.equals(code, DicDataEnum.PORTABLE.getCode()) && csEquipmentDeliveryVO.getStatus() == 1)
|| (Objects.equals(code, DicDataEnum.DEV_CLD.getCode()) && csEquipmentDeliveryVO.getStatus() == 1)){
csEquipmentDeliveryVO.setStatus(5);
} else if (DataParam.portableDevType.equals(csEquipmentDeliveryVO.getDevType()) && csEquipmentDeliveryVO.getStatus() == 2) {
} else if((Objects.equals(code, DicDataEnum.PORTABLE.getCode()) && csEquipmentDeliveryVO.getStatus() == 2)
|| (Objects.equals(code, DicDataEnum.DEV_CLD.getCode()) && csEquipmentDeliveryVO.getStatus() == 2)){
csEquipmentDeliveryVO.setStatus(6);
}
//判断装置是否已经连接上mqtt服务器
@@ -435,7 +454,7 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
List<CsDataSet> dataSet = new ArrayList<>();
//如果没有传lineId(测点ID) 则根据设备ID获取对应的模板 select modelId from cs_dev_model_relation where dev_id = ?
List<CsDevModelRelationPO> modelId = csDevModelRelationService.findModelByDevId(deviceId);
if (CollUtil.isNotEmpty(modelId)) {
if (CollUtil.isNotEmpty(modelId)){
for (CsDevModelRelationPO item : modelId) {
//再根据模板ID获取数据集 作为页面展示的Tab项 SELECT * FROM cs_data_set WHERE (pid = ? AND type IN (0,2))
dataSet.addAll(csDataSetService.findDataSetByModelId(item.getModelId()));

31
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/CsLedgerServiceImpl.java
View File

@@ -21,7 +21,6 @@ import com.njcn.csdevice.pojo.dto.LineParamDTO;
import com.njcn.csdevice.pojo.param.CsLedgerParam;
import com.njcn.csdevice.pojo.po.*;
import com.njcn.csdevice.pojo.vo.CsLedgerVO;
import com.njcn.csdevice.pojo.vo.CsMarketDataVO;
import com.njcn.csdevice.service.*;
import com.njcn.csharmonic.api.PqSensitiveUserFeignClient;
import com.njcn.device.biz.pojo.po.PqSensitiveUser;
@@ -1296,9 +1295,19 @@ public class CsLedgerServiceImpl extends ServiceImpl<CsLedgerMapper, CsLedger> i
if (CollectionUtil.isNotEmpty(enginingeringIds)) {
List<CsLedger> engineer = this.listByIds(enginingeringIds);
engineer.forEach(item -> {
List<String> devNames = new ArrayList<>();
List<String> devIds = new ArrayList<>();
DevDetailDTO detail = new DevDetailDTO();
detail.setEngineeringid(item.getId());
detail.setEngineeringName(item.getName());
ledgers.forEach(item2->{
if (item2.getPids().contains(item.getId())) {
devIds.add(item2.getId());
devNames.add(item2.getName());
}
});
detail.setEquipmentId(String.join(",", devIds));
detail.setEquipmentName(String.join(",", devNames));
details.add(detail);
});
}
@@ -1395,15 +1404,17 @@ public class CsLedgerServiceImpl extends ServiceImpl<CsLedgerMapper, CsLedger> i
LambdaQueryWrapper<CsLedger> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(CsLedger::getPid, item);
List<CsLedger> project = this.list(queryWrapper);
//工程id
List<String> projectIds = project.stream().map(CsLedger::getId).collect(Collectors.toList());
LambdaQueryWrapper<CsLedger> queryWrapper2 = new LambdaQueryWrapper<>();
queryWrapper2.in(CsLedger::getPid, projectIds);
List<CsLedger> dev = this.list(queryWrapper2);
if (CollectionUtil.isNotEmpty(dev)) {
DevDetailDTO dto = new DevDetailDTO();
dto.setEngineeringid(item);
result.add(dto);
if (CollectionUtil.isNotEmpty(project)) {
//项目id
List<String> projectIds = project.stream().map(CsLedger::getId).collect(Collectors.toList());
LambdaQueryWrapper<CsLedger> queryWrapper2 = new LambdaQueryWrapper<>();
queryWrapper2.in(CsLedger::getPid, projectIds);
List<CsLedger> dev = this.list(queryWrapper2);
if (CollectionUtil.isNotEmpty(dev)) {
DevDetailDTO dto = new DevDetailDTO();
dto.setEngineeringid(item);
result.add(dto);
}
}
});
return result;

70
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/CsTerminalLogsServiceImpl.java
View File

@@ -66,6 +66,8 @@ public class CsTerminalLogsServiceImpl extends ServiceImpl<CsTerminalLogsMapper,
wrapper.eq(CsTerminalLogs::getIsPush, 0);
List<CsTerminalLogs> list = this.list(wrapper);
if (CollectionUtil.isNotEmpty(list)) {
//先清空缓存
redisUtil.delete(RequestUtil.getUserIndex()+"reply");
//新增台账集合
List<String> addList = new ArrayList<>();
//修改台账集合
@@ -97,42 +99,44 @@ public class CsTerminalLogsServiceImpl extends ServiceImpl<CsTerminalLogsMapper,
});
//整合后 所有设备的id
List<String> devList = Stream.of(addList, updateList, deleteList).flatMap(List::stream).collect(Collectors.toList());
//获取设备集合
List<CsEquipmentDeliveryPO> deviceList = csEquipmentDeliveryService.listByIds(devList);
//按照前置机id分组
Map<String,List<CsEquipmentDeliveryPO>> nodeMap = deviceList.stream().collect(Collectors.groupingBy(CsEquipmentDeliveryPO::getNodeId));
nodeMap.forEach((k,v)->{
int maxProcessNum = nodeService.getNodeById(k).getMaxProcessNum();
//按照进程号分组
Map<Integer,List<CsEquipmentDeliveryPO>> nodeProcessMap = v.stream().collect(Collectors.groupingBy(CsEquipmentDeliveryPO::getNodeProcess));
nodeProcessMap.forEach((k1,v1)->{
if (v1.size() > 10) {
//一个进程下修改的设备数量超过10台，重启该进程号下的前置
CldControlMessage cldControlMessage = new CldControlMessage();
cldControlMessage.setGuid(IdUtil.simpleUUID());
cldControlMessage.setCode("set_process");
cldControlMessage.setProcessNo(k1);
cldControlMessage.setFun("delete");
cldControlMessage.setProcessNum(maxProcessNum);
controlMessageTemplate.sendMember(cldControlMessage,k);
}
if (CollectionUtil.isNotEmpty(devList)) {
//获取设备集合
List<CsEquipmentDeliveryPO> deviceList = csEquipmentDeliveryService.listByIds(devList);
//按照前置机id分组
Map<String,List<CsEquipmentDeliveryPO>> nodeMap = deviceList.stream().collect(Collectors.groupingBy(CsEquipmentDeliveryPO::getNodeId));
nodeMap.forEach((k,v)->{
int maxProcessNum = nodeService.getNodeById(k).getMaxProcessNum();
//按照进程号分组
Map<Integer,List<CsEquipmentDeliveryPO>> nodeProcessMap = v.stream().collect(Collectors.groupingBy(CsEquipmentDeliveryPO::getNodeProcess));
nodeProcessMap.forEach((k1,v1)->{
if (v1.size() > 10) {
//一个进程下修改的设备数量超过10台，重启该进程号下的前置
CldControlMessage cldControlMessage = new CldControlMessage();
cldControlMessage.setGuid(IdUtil.simpleUUID());
cldControlMessage.setCode("set_process");
cldControlMessage.setProcessNo(k1);
cldControlMessage.setFun("delete");
cldControlMessage.setProcessNum(maxProcessNum);
controlMessageTemplate.sendMember(cldControlMessage,k);
}
});
});
});
if (CollectionUtil.isNotEmpty(addList)) {
sendMessage(addList, deviceList, "add_terminal");
}
if (CollectionUtil.isNotEmpty(updateList)) {
sendMessage(updateList, deviceList, "ledger_modify");
}
if (CollectionUtil.isNotEmpty(deleteList)) {
sendDeleteMessage(deleteList, list, "delete_terminal");
}
if (CollectionUtil.isNotEmpty(addList)) {
sendMessage(addList, deviceList, "add_terminal");
}
if (CollectionUtil.isNotEmpty(updateList)) {
sendMessage(updateList, deviceList, "ledger_modify");
}
if (CollectionUtil.isNotEmpty(deleteList)) {
sendDeleteMessage(deleteList, list, "delete_terminal");
}
//推送完将数据改成推送
LambdaUpdateWrapper<CsTerminalLogs> wrapper2 = new LambdaUpdateWrapper<>();
wrapper2.set(CsTerminalLogs::getIsPush, 1);
this.update(wrapper2);
//推送完将数据改成推送
LambdaUpdateWrapper<CsTerminalLogs> wrapper2 = new LambdaUpdateWrapper<>();
wrapper2.set(CsTerminalLogs::getIsPush, 1);
this.update(wrapper2);
}
} else {
return "暂无需要推送的数据";
}

127
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/CsTerminalReplyServiceImpl.java
View File

@@ -57,53 +57,98 @@ public class CsTerminalReplyServiceImpl extends ServiceImpl<CsTerminalReplyMappe
List<String> redisList = Stream.of(object.toString().split(",")).collect(Collectors.toList());
List<CsTerminalReply> list = this.lambdaQuery().in(CsTerminalReply::getReplyId,redisList).orderByAsc(CsTerminalReply::getCreateTime).list();
if (CollectionUtil.isNotEmpty(list)) {
Map<Integer,List<CsTerminalReply>> map = list.stream().collect(Collectors.groupingBy(CsTerminalReply::getIsReceived));
List<CsTerminalReply> list1 = map.get(1);
if (CollectionUtil.isEmpty(list1)) {
String key = "更新失败,未收到前置应答,请查看应答报文是否发送";
result.add(key);
//将cs_terminal_logs数据置为未发送
list.forEach(item->{
list.forEach(item->{
String key;
String code = "";
if (Objects.equals(item.getCode(), "add_terminal")) {
code = "新增";
} else if (Objects.equals(item.getCode(), "ledger_modify")) {
code = "修改";
} else if (Objects.equals(item.getCode(), "delete_terminal")){
code = "删除";
}
String nodeName = nodeService.getNodeById(item.getNodeId()).getName();
List<CsEquipmentDeliveryPO> devList1 = csEquipmentDeliveryService.getAll();
List<CsEquipmentDeliveryPO> devList2 = devList1.stream().filter(item1 -> Objects.equals(item1.getId(), item.getDeviceId())).collect(Collectors.toList());
List<String> devNameList = devList2.stream().map(CsEquipmentDeliveryPO::getName).collect(Collectors.toList());
String devNameListString;
if (CollectionUtil.isNotEmpty(devNameList)) {
devNameListString = devNameList.toString();
} else {
devNameListString = "[" + item.getDeviceName() + "]";
}
if (item.getIsReceived() == 0) {
key = nodeName + item.getProcessNo() + "号进程下," + devNameListString + "设备未收到应答";
//将cs_terminal_logs数据置为未发送
csTerminalLogsService.updateLaterData(item.getDeviceId(),item.getCode());
});
} else {
list.forEach(item->{
String key;
String code = "";
if (Objects.equals(item.getCode(), "add_terminal")) {
code = "新增";
} else if (Objects.equals(item.getCode(), "ledger_modify")) {
code = "修改";
} else if (Objects.equals(item.getCode(), "delete_terminal")){
code = "删除";
}
String nodeName = nodeService.getNodeById(item.getNodeId()).getName();
List<CsEquipmentDeliveryPO> devList1 = csEquipmentDeliveryService.getAll();
List<CsEquipmentDeliveryPO> devList2 = devList1.stream().filter(item1 -> Objects.equals(item1.getId(), item.getDeviceId())).collect(Collectors.toList());
List<String> devNameList = devList2.stream().map(CsEquipmentDeliveryPO::getName).collect(Collectors.toList());
String devNameListString;
if (CollectionUtil.isNotEmpty(devNameList)) {
devNameListString = devNameList.toString();
} else {
devNameListString = "[" + item.getDeviceName() + "]";
}
if (item.getIsReceived() == 0) {
key = nodeName + item.getProcessNo() + "号进程下," + devNameListString + "设备未收到应答";
//将cs_terminal_logs数据置为未发送
csTerminalLogsService.updateLaterData(item.getDeviceId(),item.getCode());
} else if (item.getIsReceived() == 1){
key = nodeName + item.getProcessNo() + "号进程下," + devNameListString + "设备" + code + "数据成功";
} else {
key = nodeName + item.getProcessNo() + "号进程下," + devNameListString + "设备" + code + "数据失败";
}
result.add(key);
});
}
} else if (item.getIsReceived() == 1){
key = nodeName + item.getProcessNo() + "号进程下," + devNameListString + "设备" + code + "数据成功";
} else {
key = nodeName + item.getProcessNo() + "号进程下," + devNameListString + "设备" + code + "数据失败";
}
result.add(key);
});
}
}
return result;
}
// @Override
// public List<String> queryReplyData() {
// List<String> result = new ArrayList<>();
// Object object = redisUtil.getObjectByKey(RequestUtil.getUserIndex()+"reply");
// if (object != null) {
// List<String> redisList = Stream.of(object.toString().split(",")).collect(Collectors.toList());
// List<CsTerminalReply> list = this.lambdaQuery().in(CsTerminalReply::getReplyId,redisList).orderByAsc(CsTerminalReply::getCreateTime).list();
// if (CollectionUtil.isNotEmpty(list)) {
// Map<Integer,List<CsTerminalReply>> map = list.stream().collect(Collectors.groupingBy(CsTerminalReply::getIsReceived));
// List<CsTerminalReply> list1 = map.get(1);
// if (CollectionUtil.isEmpty(list1)) {
// String key = "更新失败,未收到前置应答,请查看应答报文是否发送";
// result.add(key);
// //将cs_terminal_logs数据置为未发送
// list.forEach(item->{
// csTerminalLogsService.updateLaterData(item.getDeviceId(),item.getCode());
// });
// } else {
// list.forEach(item->{
// String key;
// String code = "";
// if (Objects.equals(item.getCode(), "add_terminal")) {
// code = "新增";
// } else if (Objects.equals(item.getCode(), "ledger_modify")) {
// code = "修改";
// } else if (Objects.equals(item.getCode(), "delete_terminal")){
// code = "删除";
// }
// String nodeName = nodeService.getNodeById(item.getNodeId()).getName();
// List<CsEquipmentDeliveryPO> devList1 = csEquipmentDeliveryService.getAll();
// List<CsEquipmentDeliveryPO> devList2 = devList1.stream().filter(item1 -> Objects.equals(item1.getId(), item.getDeviceId())).collect(Collectors.toList());
// List<String> devNameList = devList2.stream().map(CsEquipmentDeliveryPO::getName).collect(Collectors.toList());
// String devNameListString;
// if (CollectionUtil.isNotEmpty(devNameList)) {
// devNameListString = devNameList.toString();
// } else {
// devNameListString = "[" + item.getDeviceName() + "]";
// }
// if (item.getIsReceived() == 0) {
// key = nodeName + item.getProcessNo() + "号进程下," + devNameListString + "设备未收到应答";
// //将cs_terminal_logs数据置为未发送
// csTerminalLogsService.updateLaterData(item.getDeviceId(),item.getCode());
// } else if (item.getIsReceived() == 1){
// key = nodeName + item.getProcessNo() + "号进程下," + devNameListString + "设备" + code + "数据成功";
// } else {
// key = nodeName + item.getProcessNo() + "号进程下," + devNameListString + "设备" + code + "数据失败";
// }
// result.add(key);
// });
// }
// }
// }
// return result;
// }
@Override
public void updateReplyData(IcdBzReplyParam param) {
LambdaUpdateWrapper<CsTerminalReply> wrapper = new LambdaUpdateWrapper<>();

4
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/DeviceFtpServiceImpl.java
View File

@@ -99,6 +99,10 @@ public class DeviceFtpServiceImpl implements DeviceFtpService {
if (!mqttClient) {
throw new BusinessException(AccessResponseEnum.MISSING_CLIENT);
}
//判断文件如果过大，不给下载
if (size > 15728640) {
throw new BusinessException("文件过大(超过15M),暂不支持下载!");
}
Object task = redisUtil.getObjectByKey("fileDowning:"+nDid);
if (Objects.nonNull(task)) {
throw new BusinessException(AlgorithmResponseEnum.FILE_DOWNLOADING);

39
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/DeviceMessageServiceImpl.java
View File

@@ -1,9 +1,17 @@
package com.njcn.csdevice.service.impl;
import cn.hutool.core.collection.CollectionUtil;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.csdevice.api.CsDeviceUserFeignClient;
import com.njcn.csdevice.param.DeviceMessageParam;
import com.njcn.csdevice.pojo.po.CsLinePO;
import com.njcn.csdevice.service.CsLinePOService;
import com.njcn.csdevice.service.DeviceMessageService;
import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.utils.RedisUtil;
import com.njcn.system.api.DicDataFeignClient;
import com.njcn.system.enums.DicDataEnum;
import com.njcn.system.pojo.po.DictData;
import com.njcn.user.api.AppInfoSetFeignClient;
import com.njcn.user.api.AppUserFeignClient;
import com.njcn.user.api.UserFeignClient;
@@ -12,8 +20,7 @@ import com.njcn.user.pojo.po.app.AppInfoSet;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.List;
import java.util.*;
import java.util.stream.Collectors;
@Service
@@ -24,6 +31,9 @@ class DeviceMessageServiceImpl implements DeviceMessageService {
private final CsDeviceUserFeignClient csDeviceUserFeignClient;
private final AppInfoSetFeignClient appInfoSetFeignClient;
private final UserFeignClient userFeignClient;
private final CsLinePOService csLinePOService;
private final DicDataFeignClient dicDataFeignClient;
private final RedisUtil redisUtil;
@Override
public List<String> getEventUserByDeviceId(String devId, Boolean isAdmin) {
@@ -74,4 +84,29 @@ class DeviceMessageServiceImpl implements DeviceMessageService {
}
return users;
}
@Override
public void getLineInfo(String id) {
Map<Integer,String> map = new HashMap<>();
List<CsLinePO> lineList = csLinePOService.findByNdid(id);
if (CollectionUtil.isEmpty(lineList)){
throw new BusinessException("监测点为空");
}
for (CsLinePO item : lineList) {
if (Objects.isNull(item.getPosition())){
map.put(item.getClDid(),item.getLineId());
} else {
DictData dictData = dicDataFeignClient.getDicDataById(item.getPosition()).getData();
if (Objects.equals(dictData.getCode(), DicDataEnum.OUTPUT_SIDE.getCode())){
map.put(0,item.getLineId());
} else if (Objects.equals(dictData.getCode(), DicDataEnum.GRID_SIDE.getCode())){
map.put(1,item.getLineId());
} else if (Objects.equals(dictData.getCode(), DicDataEnum.LOAD_SIDE.getCode())){
map.put(2,item.getLineId());
}
}
}
redisUtil.saveByKey(AppRedisKey.LINE_POSITION+id,map);
}
}

417
cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/SmsSendServiceImpl.java Normal file
View File

@@ -0,0 +1,417 @@
package com.njcn.csdevice.service.impl;
import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.alibaba.nacos.shaded.com.google.gson.JsonObject;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.csdevice.mapper.CsSmsSendRecordMapper;
import com.njcn.csdevice.pojo.dto.CredentialReqDTO;
import com.njcn.csdevice.pojo.dto.SendResult;
import com.njcn.csdevice.pojo.po.CsSmsSendRecord;
import com.njcn.csdevice.pojo.vo.MessageRecordReqVO;
import com.njcn.csdevice.service.ISmsSendService;
import com.njcn.redis.utils.RedisUtil;
import lombok.AllArgsConstructor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.HttpURLConnection;
import java.net.SocketTimeoutException;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
/**
* @author xy
*/
@Slf4j
@Service
@AllArgsConstructor
@RequiredArgsConstructor
public class SmsSendServiceImpl extends ServiceImpl<CsSmsSendRecordMapper, CsSmsSendRecord> implements ISmsSendService {
@Value("${msg.credential_url:http://192.168.2.126:48083/admin-api/push/credential/generate}")
private String CREDENTIAL_URL;
@Value("${msg.sms_send_url:http://192.168.2.126:48083/admin-api/push/message/send/sms}")
private String SMS_SEND_URL;
@Value("${msg.connect_timeout:5000}")
private Integer CONNECT_TIMEOUT;
@Value("${msg.read_timeout:30000}")
private Integer READ_TIMEOUT;
@Value("${msg.system_name:NPQS-9500}")
private String SYSTEM_NAME;
@Value("${msg.secret_key:123456}")
private String SECRET_KEY;
private static final int[] RETRY_DELAYS = {1, 2, 3};
private static final int MAX_RETRY = 3;
private static final String CREDENTIAL_CACHE_KEY = "SMS_CREDENTIAL_TOKEN";
private static final Gson GSON = new Gson();
@Resource
private RedisUtil redisUtil;
@Override
public void sendSmsWithRetry(String receiver, String content, String messageType) {
MessageRecordReqVO vo = new MessageRecordReqVO();
vo.setReceiver(receiver);
vo.setContent(content);
vo.setMessageType(messageType);
sendSmsWithRetry(vo);
}
@Override
public void sendSmsWithRetry(MessageRecordReqVO messageRecordReqVO) {
CsSmsSendRecord record = initRecord(messageRecordReqVO);
this.save(record);
try {
String credentialToken = getOrRefreshCredentialWithRetry(record);
if (credentialToken == null) {
record.setSendStatus(0);
record.setFailReason("获取凭证失败，已重试3次");
log.error("获取凭证失败，短信未发送，接收者: {}", messageRecordReqVO.getReceiver());
this.updateById(record);
throw new BusinessException("获取凭证失败，已重试3次");
}
record.setCredentialToken(credentialToken);
record.setSendTime(LocalDateTime.now());
this.updateById(record);
boolean success = attemptSendWithRetry(messageRecordReqVO, credentialToken, record);
if (success) {
record.setSendStatus(1);
record.setFailReason(null);
log.info("短信发送成功，接收者: {}", messageRecordReqVO.getReceiver());
} else {
record.setSendStatus(0);
if (record.getFailReason() == null) {
record.setFailReason("超过最大重试次数，发送失败");
}
log.error("短信发送失败，接收者: {}，已重试{}次，原因: {}",
messageRecordReqVO.getReceiver(), record.getRetryCount(), record.getFailReason());
throw new BusinessException("短信发送失败: " + record.getFailReason());
}
} catch (BusinessException e) {
record.setSendStatus(0);
record.setFailReason(e.getMessage());
log.error("短信发送业务异常，接收者: {}", messageRecordReqVO.getReceiver(), e);
this.updateById(record);
throw e;
} catch (Exception e) {
record.setSendStatus(0);
record.setFailReason("发送异常: " + e.getMessage());
log.error("短信发送异常，接收者: {}", messageRecordReqVO.getReceiver(), e);
this.updateById(record);
throw new BusinessException("短信发送异常: " + e.getMessage());
} finally {
this.updateById(record);
}
}
private CsSmsSendRecord initRecord(MessageRecordReqVO vo) {
CsSmsSendRecord record = new CsSmsSendRecord();
record.setReceiver(vo.getReceiver());
record.setContent(vo.getContent());
record.setMessageType(vo.getMessageType());
record.setSendStatus(-1);
record.setRetryCount(0);
record.setMaxRetry(MAX_RETRY);
record.setCreateTime(LocalDateTime.now());
return record;
}
private String getOrRefreshCredentialWithRetry(CsSmsSendRecord record) {
Object cachedToken = redisUtil.getObjectByKey(CREDENTIAL_CACHE_KEY);
if (cachedToken != null) {
log.info("使用缓存的凭证令牌");
return cachedToken.toString();
}
log.info("缓存中无凭证，开始获取新凭证（最多重试3次）");
for (int i = 1; i <= 3; i++) {
try {
String token = fetchNewCredential();
log.info("第{}次尝试获取凭证成功", i);
return token;
} catch (Exception e) {
log.warn("第{}次获取凭证失败: {}", i, e.getMessage());
record.setFailReason("获取凭证第" + i + "次失败: " + e.getMessage());
this.updateById(record);
try {
int waitSeconds = i * 10;
log.info("等待{}秒后重试...", waitSeconds);
TimeUnit.SECONDS.sleep(waitSeconds);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
log.error("凭证获取重试被中断");
record.setFailReason("获取凭证实例被中断");
this.updateById(record);
return null;
}
}
}
log.error("获取凭证失败，已重试3次");
return null;
}
private String fetchNewCredential() {
CredentialReqDTO reqDTO = new CredentialReqDTO();
reqDTO.setSystemName(SYSTEM_NAME);
reqDTO.setSecretKey(SECRET_KEY);
HttpURLConnection connection = null;
try {
URL url = new URL(CREDENTIAL_URL);
connection = (HttpURLConnection) url.openConnection();
connection.setRequestMethod("POST");
connection.setRequestProperty("Content-Type", "application/json");
connection.setConnectTimeout(CONNECT_TIMEOUT);
connection.setReadTimeout(READ_TIMEOUT);
connection.setDoOutput(true);
OutputStream outputStream = connection.getOutputStream();
outputStream.write(GSON.toJson(reqDTO).getBytes(StandardCharsets.UTF_8));
outputStream.flush();
outputStream.close();
int responseCode = connection.getResponseCode();
if (responseCode != 200) {
throw new BusinessException("获取凭证失败，HTTP响应码: " + responseCode);
}
BufferedReader reader = new BufferedReader(
new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8));
StringBuilder response = new StringBuilder();
String inputLine;
while ((inputLine = reader.readLine()) != null) {
response.append(inputLine);
}
reader.close();
JsonObject jsonResponse = GSON.fromJson(response.toString(), JsonObject.class);
int code = jsonResponse.get("code").getAsInt();
if (code != 0) {
String msg = jsonResponse.has("msg") ? jsonResponse.get("msg").getAsString() : "未知错误";
throw new BusinessException("获取凭证失败，错误码: " + code + "，错误信息: " + msg);
}
JsonObject data = jsonResponse.getAsJsonObject("data");
String token = data.get("credentialToken").getAsString();
long expiresTimestamp = data.get("expiresTime").getAsLong();
LocalDateTime expiresTime = LocalDateTime.ofInstant(
Instant.ofEpochMilli(expiresTimestamp),
ZoneId.systemDefault()
);
long expireSeconds = calculateExpireSeconds(expiresTime);
redisUtil.saveByKeyWithExpire(CREDENTIAL_CACHE_KEY, token, expireSeconds);
log.info("获取新凭证成功，过期时间: {}，缓存有效期: {}秒", expiresTime, expireSeconds);
return token;
} catch (SocketTimeoutException e) {
throw new BusinessException("获取凭证超时(30秒)，请检查网络连接");
} catch (ConnectException e) {
throw new BusinessException("无法连接到凭证服务，请检查服务是否启动和网络是否正常");
} catch (IOException e) {
throw new BusinessException("获取凭证IO异常: " + e.getMessage());
} finally {
if (connection != null) {
connection.disconnect();
}
}
}
private long calculateExpireSeconds(LocalDateTime expiresTime) {
long expireSeconds = java.time.Duration.between(
LocalDateTime.now(),
expiresTime
).getSeconds();
expireSeconds = expireSeconds - 60;
return Math.max(expireSeconds, 60);
}
private boolean attemptSendWithRetry(MessageRecordReqVO vo, String token, CsSmsSendRecord record) {
for (int attempt = 0; attempt <= MAX_RETRY; attempt++) {
if (attempt > 0) {
int delayMinutes = RETRY_DELAYS[attempt - 1];
log.info("第{}次重试，等待{}分钟后发送...", attempt, delayMinutes);
try {
TimeUnit.MINUTES.sleep(delayMinutes);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.error("重试等待被中断", e);
record.setFailReason("重试等待被中断");
return false;
}
record.setRetryCount(attempt);
}
SendResult result = executeSendSms(vo, token, record);
if (result.isSuccess()) {
record.setFailReason(null);
log.info("第{}次尝试发送成功，消息ID: {}", attempt, result.getMessageId());
return true;
}
record.setFailReason(result.getFailReason());
if (result.isUnauthorized()) {
log.warn("凭证失效(401)，重新获取凭证后重试...");
String newToken = getOrRefreshCredentialWithRetry(record);
if (newToken == null) {
record.setFailReason("凭证刷新失败，无法重新获取凭证");
return false;
}
token = newToken;
record.setCredentialToken(newToken);
this.updateById(record);
continue;
}
if (!result.isTimeOut()) {
log.warn("发送失败且非超时，不再重试，原因: {}，响应时间: {}ms",
result.getFailReason(), record.getResponseTime());
return false;
}
log.warn("第{}次发送超时，将重试，响应时间: {}ms",
attempt, record.getResponseTime());
}
record.setFailReason("超过最大重试次数，发送超时");
return false;
}
private SendResult executeSendSms(MessageRecordReqVO vo, String token, CsSmsSendRecord record) {
HttpURLConnection connection = null;
long startTime = System.currentTimeMillis();
try {
URL url = new URL(SMS_SEND_URL);
connection = (HttpURLConnection) url.openConnection();
connection.setRequestMethod("POST");
connection.setRequestProperty("Content-Type", "application/json");
connection.setRequestProperty("X-Credential-Token", token);
connection.setConnectTimeout(CONNECT_TIMEOUT);
connection.setReadTimeout(READ_TIMEOUT);
connection.setDoOutput(true);
OutputStream outputStream = connection.getOutputStream();
outputStream.write(GSON.toJson(Collections.singletonList(vo)).getBytes(StandardCharsets.UTF_8));
outputStream.flush();
outputStream.close();
int responseCode = connection.getResponseCode();
long responseTime = System.currentTimeMillis() - startTime;
record.setResponseTime(responseTime);
BufferedReader reader = new BufferedReader(
new InputStreamReader(
responseCode == 200 ? connection.getInputStream() : connection.getErrorStream(),
StandardCharsets.UTF_8
)
);
StringBuilder response = new StringBuilder();
String inputLine;
while ((inputLine = reader.readLine()) != null) {
response.append(inputLine);
}
reader.close();
if (responseCode != 200) {
String failReason = "HTTP响应码异常: " + responseCode;
if (response.length() > 0) {
failReason += "，响应: " + response.toString();
}
record.setFailReason(failReason);
return new SendResult(false, null, failReason, false, false);
}
JsonObject jsonResponse = GSON.fromJson(response.toString(), JsonObject.class);
int code = jsonResponse.get("code").getAsInt();
if (code == 401) {
String failReason = "凭证失效(HTTP 401)";
record.setFailReason(failReason);
redisUtil.delete(CREDENTIAL_CACHE_KEY);
return new SendResult(false, null, failReason, false, true);
} else {
if (code != 0) {
String msg = jsonResponse.has("msg") ? jsonResponse.get("msg").getAsString() : "未知错误";
String failReason = "业务错误码: " + code + "，错误信息: " + msg;
record.setFailReason(failReason);
return new SendResult(false, null, failReason, false, false);
}
}
JsonObject firstResult = jsonResponse.getAsJsonArray("data").get(0).getAsJsonObject();
boolean result = firstResult.get("result").getAsBoolean();
String messageId = firstResult.has("messageId") ? firstResult.get("messageId").getAsString() : null;
String detail = firstResult.has("detail") ? firstResult.get("detail").getAsString() : null;
if (result) {
log.info("短信发送成功，接收者: {}，消息ID: {}，详情: {}，耗时: {}ms",
vo.getReceiver(), messageId, detail, responseTime);
return new SendResult(true, messageId, null, false, false);
} else {
String failReason = "发送失败: " + detail;
record.setFailReason(failReason);
return new SendResult(false, messageId, failReason, false, false);
}
} catch (SocketTimeoutException e) {
long responseTime = System.currentTimeMillis() - startTime;
record.setResponseTime(responseTime);
String failReason = "请求超时(30秒)";
record.setFailReason(failReason);
log.warn("短信发送超时，接收者: {}，耗时: {}ms", vo.getReceiver(), responseTime);
return new SendResult(false, null, failReason, true, false);
} catch (ConnectException e) {
long responseTime = System.currentTimeMillis() - startTime;
record.setResponseTime(responseTime);
String failReason = "无法连接到短信服务";
record.setFailReason(failReason);
log.error("短信服务连接失败，接收者: {}", vo.getReceiver(), e);
return new SendResult(false, null, failReason, false, false);
} catch (IOException e) {
long responseTime = System.currentTimeMillis() - startTime;
record.setResponseTime(responseTime);
String failReason = "IO异常: " + e.getMessage();
record.setFailReason(failReason);
log.error("短信发送IO异常，接收者: {}，耗时: {}ms", vo.getReceiver(), responseTime, e);
return new SendResult(false, null, failReason, false, false);
} finally {
if (connection != null) {
connection.disconnect();
}
}
}
}