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

fix: 电话报警通知逻辑修改。

Browse Source
This commit is contained in:
tianyongbao
2026-03-07 22:13:40 +08:00
parent 212e596237
commit 10cc7e75b3
14 changed files with 557 additions and 1294 deletions
Download Patch File Download Diff File Expand all files Collapse all files

285
intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/VmsMnsConsumerServiceImpl.java
View File

@@ -5,6 +5,11 @@ import com.aliyun.mns.client.CloudAccount;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.model.Message;
import com.aliyuncs.DefaultAcsClient;
import com.aliyuncs.IAcsClient;
import com.aliyuncs.dybaseapi.model.v20170525.QueryTokenForMnsQueueRequest;
import com.aliyuncs.dybaseapi.model.v20170525.QueryTokenForMnsQueueResponse;
import com.aliyuncs.profile.DefaultProfile;
import com.intc.iot.config.AliyunIotProperties;
import com.intc.iot.domain.VmsCallback;
import com.intc.iot.mapper.VmsCallbackMapper;
@@ -17,15 +22,18 @@ import org.springframework.stereotype.Service;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
/**
* VMS MNS 回执消费服务实现
* 注意：直接使用 AccessKey/Secret 访问 MNS，不再依赖 dybaseapi
* 使用 STS 临时凭证（QueryTokenForMnsQueue）动态获取 Token，对应 C# 的 MNSHandle 实现逻辑
*
* @author intc-iot
*/
@@ -39,66 +47,72 @@ public class VmsMnsConsumerServiceImpl implements VmsMnsConsumerService {
private final AliyunIotProperties aliyunIotProperties;
private final VmsCallbackMapper vmsCallbackMapper;
/**
* MNS 配置常量
* 注意：这些值需要根据你的实际阿里云账户配置
*/
private static final String VMS_MNS_QUEUE_NAME = "Alicom-Queue-1572610294777992-VoiceReport";
private static final String MNS_ACCOUNT_ENDPOINT = "https://1943695596114318.mns.cn-hangzhou.aliyuncs.com/";
/** MNS 消息类型（固定值） */
private static final String VMS_MNS_MESSAGE_TYPE = "VoiceReport";
/** MNS 默认线程数 */
private static final int MNS_THREAD_COUNT = 2;
private static final int POLL_WAIT_SECONDS = 30;
/**
* 应用标识（用于过滤 out_id）
*/
private String appName;
/** Token 过期缓冲时间（120秒），小于此值则重新获取，防止服务器时间差异 */
private static final long TOKEN_BUFFER_SECONDS = 120;
/**
* MNS 客户端
*/
/** 线程安全的 Token 缓存 */
private final ReentrantLock tokenLock = new ReentrantLock();
private QueryTokenForMnsQueueResponse.MessageTokenDTO cachedToken;
private CloudQueue cachedQueue;
/** ACS 客户端（用于获取 STS Token） */
private IAcsClient acsClient;
/** MNS 客户端（临时凭证建立） */
private MNSClient mnsClient;
/**
* 线程池
*/
/** 线程池 */
private ExecutorService executorService;
/**
* 运行状态标志
*/
/** 运行状态标志 */
private final AtomicBoolean running = new AtomicBoolean(false);
@Override
public void start() throws Exception {
if (running.compareAndSet(false, true)) {
log.info("启动 VMS MNS 回执消费服务...");
log.info("启动 VMS MNS 回执消费服务（STS 临时凭证模式）...");
// 从配置读取应用名称
appName = aliyunIotProperties.getAppKey() != null && !aliyunIotProperties.getAppKey().isEmpty()
? aliyunIotProperties.getAppKey()
: "fishery-backend";
log.info("VMS 回执过滤应用名称: {}", appName);
AliyunIotProperties.VmsMnsConfig vmsMns = aliyunIotProperties.getVms();
String mnsEndpoint = vmsMns.getEndpoint();
String mnsQueueName = vmsMns.getQueueName();
// 直接使用 AccessKey/Secret 创建 MNS 客户端
CloudAccount account = new CloudAccount(
aliyunIotProperties.getAccessKeyId(),
aliyunIotProperties.getAccessKeySecret(),
MNS_ACCOUNT_ENDPOINT
);
mnsClient = account.getMNSClient();
log.info("MNS 客户端初始化成功，Endpoint: {}", MNS_ACCOUNT_ENDPOINT);
if (mnsEndpoint == null || mnsEndpoint.isEmpty()) {
log.error("VMS MNS Endpoint 未配置，请在 aliyun.living-iot.vms.endpoint 中配置");
running.set(false);
return;
}
// 使用主 AK 创建 ACS 客户端（用于调用 QueryTokenForMnsQueue 接口）
String akId = vmsMns.getAccessKeyId() != null && !vmsMns.getAccessKeyId().isEmpty()
? vmsMns.getAccessKeyId() : aliyunIotProperties.getAccessKeyId();
String akSecret = vmsMns.getAccessKeySecret() != null && !vmsMns.getAccessKeySecret().isEmpty()
? vmsMns.getAccessKeySecret() : aliyunIotProperties.getAccessKeySecret();
String regionId = aliyunIotProperties.getRegionId() != null
? aliyunIotProperties.getRegionId() : "cn-hangzhou";
DefaultProfile profile = DefaultProfile.getProfile(regionId, akId, akSecret);
DefaultProfile.addEndpoint(regionId, regionId, "Dybaseapi", "dybaseapi.aliyuncs.com");
acsClient = new DefaultAcsClient(profile);
log.info("ACS 客户端创建成功，region: {}", regionId);
// 创建线程池
executorService = Executors.newFixedThreadPool(MNS_THREAD_COUNT);
// 启动消费线程（对应 C# 的 MNSHandle 循环）
// 启动消费线程
for (int i = 0; i < MNS_THREAD_COUNT; i++) {
executorService.submit(this::mnsConsumeLoop);
executorService.submit(() -> mnsConsumeLoop(mnsEndpoint, mnsQueueName));
}
log.info("VMS MNS 回执消费服务启动成功，线程数: {}", MNS_THREAD_COUNT);
log.info("VMS MNS 回扇消费服务启动成功，线程数: {}", MNS_THREAD_COUNT);
} else {
log.warn("VMS MNS 回执消费服务已在运行中，无需重复启动");
log.warn("VMS MNS 回扇消费服务已在运行中，无需重复启动");
}
}
@@ -119,9 +133,9 @@ public class VmsMnsConsumerServiceImpl implements VmsMnsConsumerService {
}
}
// 关闭 MNS 客户端
if (mnsClient != null) {
mnsClient.close();
mnsClient = null;
}
log.info("VMS MNS 回执消费服务已停止");
@@ -134,23 +148,93 @@ public class VmsMnsConsumerServiceImpl implements VmsMnsConsumerService {
}
/**
* MNS 消费循环
* 获取或刷新 STS Token 并返回对应的 CloudQueue
* 对应 C# 的 Token 过期检查 + 自动刷新逻辑
*
* @param mnsEndpoint MNS endpoint
* @param queueName 队列名称
* @return 可用的 CloudQueue
*/
private void mnsConsumeLoop() {
log.info("MNS 消费线程 {} 启动", Thread.currentThread().getName());
CloudQueue queue = null;
private CloudQueue getOrRefreshQueue(String mnsEndpoint, String queueName) throws Exception {
tokenLock.lock();
try {
// 获取队列引用
queue = mnsClient.getQueueRef(VMS_MNS_QUEUE_NAME);
log.info("获取 MNS Queue 成功: {}", VMS_MNS_QUEUE_NAME);
} catch (Exception e) {
log.error("获取 MNS Queue 失败: {}", e.getMessage(), e);
return;
// 检查 Token 是否将要过期（小于缓冲时间则重新获取）
if (cachedToken != null && cachedQueue != null) {
long remainSeconds = getTokenRemainSeconds(cachedToken.getExpireTime());
if (remainSeconds >= TOKEN_BUFFER_SECONDS) {
return cachedQueue;
}
log.info("STS Token 将于 {}s 内过期，重新获取...", remainSeconds);
} else {
log.info("首次获取 MNS STS Token...");
}
// 调用 QueryTokenForMnsQueue 获取临时凭证
QueryTokenForMnsQueueRequest request = new QueryTokenForMnsQueueRequest();
request.setMessageType(VMS_MNS_MESSAGE_TYPE);
QueryTokenForMnsQueueResponse response = acsClient.getAcsResponse(request);
QueryTokenForMnsQueueResponse.MessageTokenDTO token =
response.getMessageTokenDTO();
log.info("STS Token 获取成功，过期时间: {}", token.getExpireTime());
// 关闭旧的 MNS 客户端
if (mnsClient != null) {
mnsClient.close();
}
// 使用 STS 临时凭证创建新的 MNS 客户端
CloudAccount account = new CloudAccount(
token.getAccessKeyId(),
token.getAccessKeySecret(),
mnsEndpoint,
token.getSecurityToken()
);
mnsClient = account.getMNSClient();
cachedQueue = mnsClient.getQueueRef(queueName);
cachedToken = token;
log.info("MNS 客户端刷新成功，Queue: {}", queueName);
return cachedQueue;
} finally {
tokenLock.unlock();
}
}
/**
* 计算 Token 剩余有效秒数
* SDK 返回的 expireTime 格式为 "yyyy-MM-dd HH:mm:ss"，时区为 UTC+8
*/
private long getTokenRemainSeconds(String expireTime) {
try {
java.time.LocalDateTime expire = java.time.LocalDateTime.parse(
expireTime,
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
);
// SDK 返回的时间为 UTC+8
long expireEpoch = expire.toEpochSecond(java.time.ZoneOffset.ofHours(8));
return expireEpoch - System.currentTimeMillis() / 1000;
} catch (Exception e) {
log.warn("解析 Token 过期时间失败: {}", expireTime);
return 0;
}
}
/**
* MNS 消费循环（每个线程执行）
*
* @param mnsEndpoint MNS endpoint
* @param queueName 队列名称
*/
private void mnsConsumeLoop(String mnsEndpoint, String queueName) {
log.info("MNS 消费线程 {} 启动", Thread.currentThread().getName());
while (running.get()) {
try {
// 获取（或刷新）有效的 CloudQueue
CloudQueue queue = getOrRefreshQueue(mnsEndpoint, queueName);
// 批量接收消息
List<Message> messages = queue.batchPopMessage(16, POLL_WAIT_SECONDS);
@@ -160,7 +244,6 @@ public class VmsMnsConsumerServiceImpl implements VmsMnsConsumerService {
log.debug("接收到 {} 条 VMS 回执消息", messages.size());
// 处理消息
for (Message message : messages) {
try {
processMessage(message, queue);
@@ -170,14 +253,28 @@ public class VmsMnsConsumerServiceImpl implements VmsMnsConsumerService {
}
} catch (Exception e) {
// 检查是否为中断异常
if (e instanceof InterruptedException || e.getCause() instanceof InterruptedException) {
Thread.currentThread().interrupt();
log.info("MNS 消费线程被中断");
break;
}
log.error("MNS 消费循环异常: {}", e.getMessage(), e);
// MessageNotExist 表示队列为空，属于正常现象，静默处理（使用长轮询时不应出现，出现时不报错）
String errMsg = e.getMessage();
if (errMsg != null && errMsg.contains("MessageNotExist")) {
log.debug("MNS 队列暂无消息");
continue;
}
log.error("MNS 消费循环异常: {}", errMsg, e);
// 异常后清除缓存，下次循环重新获取 Token
tokenLock.lock();
try {
cachedToken = null;
cachedQueue = null;
} finally {
tokenLock.unlock();
}
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException ie) {
@@ -192,64 +289,59 @@ public class VmsMnsConsumerServiceImpl implements VmsMnsConsumerService {
/**
* 处理单条消息
*
* @param message MNS 消息
* @param queue MNS 队列
* 注意：MNS 回执消息体可能是明文 JSON，也可能是 Base64 编码的 JSON，需要尝试两种解析方式
*/
private void processMessage(Message message, CloudQueue queue) {
try {
// 解码消息体（对应 C# 的 Base64 解码）
String body = message.getMessageBodyAsString();
byte[] decoded = Base64.getDecoder().decode(body);
String jsonStr = new String(decoded, StandardCharsets.UTF_8);
// 尝试 Base64 解码，如果失败则直接当明文 JSON 处理
String jsonStr;
try {
byte[] decoded = Base64.getDecoder().decode(body.trim());
jsonStr = new String(decoded, StandardCharsets.UTF_8);
} catch (IllegalArgumentException e) {
// 消息体本身就是 JSON 明文，无需 Base64 解码
jsonStr = body;
}
log.debug("收到 VMS 回执消息: {}", jsonStr);
// 解析为实体（对应 C# 的 JSON.Deserialize<AliVmsCallBack>）
Map<String, Object> callbackMap = JSONUtil.toBean(jsonStr, Map.class);
String outId = (String) callbackMap.get("out_id");
// 过滤不属于当前系统的消息（对应 C# 的 out_id 判断）
if (outId == null || outId.isEmpty()) {
log.debug("跳过无效 out_id 的消息");
queue.deleteMessage(message.getReceiptHandle());
return;
}
// 验证 out_id 是否属于本系统（对应 C# 的 out_id.Equals(DefineValue.AppName)）
if (!outId.equals(appName)) {
log.debug("收到语音通知回执不属于当前系统, outId={}, 期望={}", outId, appName);
queue.deleteMessage(message.getReceiptHandle());
// 验证 out_id 是否属于本系统（out_id 格式为 "ALARM_{callNoticeId}_{timestamp}"）
if (!outId.startsWith("ALARM_")) {
log.debug("收到语音通知回执不属于当前系统, outId={}", outId);
// 与 C# 一致：不属于本系统的消息不删除，保留在队列
return;
}
// 构建实体
VmsCallback vmsCallback = buildVmsCallback(callbackMap);
// 入库（对应 C# 的 CacheData.sListVmsCallback.Add）
try {
vmsCallbackMapper.insert(vmsCallback);
log.info("VMS 回执入库成功 - CallId: {}, StatusCode: {}", vmsCallback.getCallId(), vmsCallback.getStatusCode());
// 成功处理后删除消息
queue.deleteMessage(message.getReceiptHandle());
} catch (org.springframework.dao.DuplicateKeyException e) {
log.warn("VMS 回执已存在，跳过 - CallId: {}", vmsCallback.getCallId());
// 重复数据也删除消息
queue.deleteMessage(message.getReceiptHandle());
}
} catch (Exception e) {
log.error("处理 VMS 回执消息异常: {}", e.getMessage(), e);
// 注意：不删除消息，让它重新回到队列，等待下次处理
// 不删除消息，等待重试
}
}
/**
* 构建 VmsCallback 实体
*
* @param map 回执 Map
* @return VmsCallback
*/
private VmsCallback buildVmsCallback(Map<String, Object> map) {
VmsCallback callback = new VmsCallback();
@@ -260,56 +352,29 @@ public class VmsMnsConsumerServiceImpl implements VmsMnsConsumerService {
callback.setHangupDirection((String) map.get("hangup_direction"));
callback.setCaller((String) map.get("caller"));
callback.setVoiceType((String) map.get("voice_type"));
// 时间戳字段
callback.setOriginateTime(getLongValue(map, "originate_time"));
callback.setStartTime(getLongValue(map, "start_time"));
callback.setEndTime(getLongValue(map, "end_time"));
// 整数字段
callback.setRingTime(getIntValue(map, "ring_time"));
callback.setDuration(getIntValue(map, "duration"));
callback.setTollType((String) map.get("toll_type"));
callback.setProcessed(0); // 初始为未处理
callback.setProcessed(0);
callback.setCreatedTime(LocalDateTime.now());
return callback;
}
/**
* 从 Map 中获取 Long 值
*/
private Long getLongValue(Map<String, Object> map, String key) {
Object value = map.get(key);
if (value == null) {
return null;
}
if (value instanceof Number) {
return ((Number) value).longValue();
}
try {
return Long.parseLong(value.toString());
} catch (NumberFormatException e) {
return null;
}
if (value == null) return null;
if (value instanceof Number) return ((Number) value).longValue();
try { return Long.parseLong(value.toString()); } catch (NumberFormatException e) { return null; }
}
/**
* 从 Map 中获取 Integer 值
*/
private Integer getIntValue(Map<String, Object> map, String key) {
Object value = map.get(key);
if (value == null) {
return null;
}
if (value instanceof Number) {
return ((Number) value).intValue();
}
try {
return Integer.parseInt(value.toString());
} catch (NumberFormatException e) {
return null;
}
if (value == null) return null;
if (value instanceof Number) return ((Number) value).intValue();
try { return Integer.parseInt(value.toString()); } catch (NumberFormatException e) { return null; }
}
}

12
intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/VmsNoticeServiceImpl.java
View File

@@ -30,10 +30,10 @@ public class VmsNoticeServiceImpl implements VmsNoticeService {
private static final String CALL_SHOW_NUMBER = "0571000013978";
/** 默认 TTS 模板编码（当前取 C# 版本中的正式模板编码，可按环境调整） */
private static final String TTS_CODE = "TTS_299550017";
private static final String TTS_CODE = "TTS_328610196";
private final AliyunIotProperties aliyunIotProperties;
/**
* VMS 客户端单例，避免重复创建
*/
@@ -68,7 +68,7 @@ public class VmsNoticeServiceImpl implements VmsNoticeService {
request.setSysDomain("dyvmsapi.aliyuncs.com");
request.setSysVersion("2017-05-25");
request.setSysAction("SingleCallByTts");
// 设置请求参数
request.putQueryParameter("CalledNumber", phoneNum);
request.putQueryParameter("CalledShowNumber", CALL_SHOW_NUMBER);
@@ -83,7 +83,7 @@ public class VmsNoticeServiceImpl implements VmsNoticeService {
try {
CommonResponse response = client.getCommonResponse(request);
if (response == null || response.getData() == null) {
log.error("[VmsNoticeService] 语音呼叫请求异常, response 为空");
VmsNoticeResponse result = new VmsNoticeResponse();
@@ -95,7 +95,7 @@ public class VmsNoticeServiceImpl implements VmsNoticeService {
// 解析响应
String responseData = response.getData();
cn.hutool.json.JSONObject jsonResponse = JSONUtil.parseObj(responseData);
String code = jsonResponse.getStr("Code");
String message = jsonResponse.getStr("Message");
String callId = jsonResponse.getStr("CallId");
@@ -107,7 +107,7 @@ public class VmsNoticeServiceImpl implements VmsNoticeService {
result.setSuccess("OK".equalsIgnoreCase(code));
if (!result.isSuccess()) {
log.warn("[VmsNoticeService] 语音呼叫失败, phoneNum={}, code={}, message={}, callId={}",
log.warn("[VmsNoticeService] 语音呼叫失败, phoneNum={}, code={}, message={}, callId={}",
phoneNum, code, message, callId);
}

46
intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/WarnCallNoticeServiceImpl.java
View File

@@ -25,7 +25,6 @@ import java.util.List;
*/
@Service
@RequiredArgsConstructor
@ConditionalOnBean({AquWarnCallNoticeMapper.class, VmsCallbackMapper.class})
@Slf4j
public class WarnCallNoticeServiceImpl implements WarnCallNoticeService {
@@ -61,17 +60,44 @@ public class WarnCallNoticeServiceImpl implements WarnCallNoticeService {
return false;
}
log.debug("[告警通知] 处理回执信息 - 呼叫ID: {}, 状态码: {}", callback.getCallId(), callback.getStatusCode());
log.debug("[告警通知] 处理回执信息 - 呼叫ID: {}, outId: {}, 状态码: {}",
callback.getCallId(), callback.getOutId(), callback.getStatusCode());
// 根据 CallId 查找对应的通知记录
AquWarnCallNotice callNotice = warnCallNoticeMapper.selectOne(
new LambdaQueryWrapper<AquWarnCallNotice>()
.eq(AquWarnCallNotice::getCallId, callback.getCallId())
.eq(AquWarnCallNotice::getCallStatus, CALL_STATUS_CALL_SUCCESS) // 只处理呼叫成功等待反馈的记录
);
// 优先从 outId（格式：ALARM_{callNoticeId}_{timestamp}）中解析主键直接查询
// 若解析失败则回退到 callId 匹配
AquWarnCallNotice callNotice = null;
String outId = callback.getOutId();
if (outId != null && outId.startsWith("ALARM_")) {
try {
// outId 格式：ALARM_{id}_{timestamp}，取第二段
String[] parts = outId.split("_");
if (parts.length >= 2) {
Long noticeId = Long.parseLong(parts[1]);
callNotice = warnCallNoticeMapper.selectOne(
new LambdaQueryWrapper<AquWarnCallNotice>()
.eq(AquWarnCallNotice::getId, noticeId)
.eq(AquWarnCallNotice::getCallStatus, CALL_STATUS_CALL_SUCCESS)
);
if (callNotice != null) {
log.debug("[告警通知] 通过 outId 找到通知记录 - 记录ID: {}", noticeId);
}
}
} catch (NumberFormatException e) {
log.warn("[告警通知] outId 解析失败，回退到 callId 匹配 - outId: {}", outId);
}
}
// 回退：通过 callId 匹配
if (callNotice == null) {
callNotice = warnCallNoticeMapper.selectOne(
new LambdaQueryWrapper<AquWarnCallNotice>()
.eq(AquWarnCallNotice::getCallId, callback.getCallId())
.eq(AquWarnCallNotice::getCallStatus, CALL_STATUS_CALL_SUCCESS)
);
}
if (callNotice == null) {
log.warn("[告警通知] 未找到对应的通知记录 - 呼叫ID: {}", callback.getCallId());
log.warn("[告警通知] 未找到对应的通知记录 - 呼叫ID: {}, outId: {}", callback.getCallId(), outId);
return false;
}
@@ -183,7 +209,6 @@ public class WarnCallNoticeServiceImpl implements WarnCallNoticeService {
// 更新状态为回执失败
notice.setCallStatus(CALL_STATUS_CALLBACK_FAIL);
notice.setStatusMsg("回执超时，未在规定时间内收到回执");
notice.setUpdatedTime(now);
warnCallNoticeMapper.updateById(notice);
cleanedCount++;
@@ -222,7 +247,6 @@ public class WarnCallNoticeServiceImpl implements WarnCallNoticeService {
callNotice.setStatusMsg(callback.getStatusMsg());
callNotice.setOutId(callback.getOutId());
callNotice.setTollType(callback.getTollType());
callNotice.setUpdatedTime(now);
// 根据状态码判断通话结果（对应 C# 的状态码判断逻辑）
if (VMS_STATUS_CODE_SUCCESS.equals(callback.getStatusCode()) ||