|
|
@@ -2,22 +2,22 @@ package com.hdkj.lt.bf.listener;
|
|
|
|
|
|
import cn.hutool.core.lang.Snowflake;
|
|
|
import cn.hutool.core.util.IdUtil;
|
|
|
+import com.alibaba.fastjson.JSON;
|
|
|
import com.alibaba.fastjson.JSONObject;
|
|
|
-import com.alibaba.fastjson2.JSON;
|
|
|
import com.fasterxml.jackson.core.JsonProcessingException;
|
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
import com.hdkj.hussar.ApiResponse;
|
|
|
-import com.hdkj.lt.base.exception.ExceptionCast;
|
|
|
import com.hdkj.lt.bf.entity.FhzgSeCurrentEvent;
|
|
|
import com.hdkj.lt.bf.entity.FhzgSeEventStateEstimation;
|
|
|
import com.hdkj.lt.bf.entity.FhzgSeVoltageEvent;
|
|
|
-import com.hdkj.lt.bf.entity.dto.XlRelDTO;
|
|
|
+import com.hdkj.lt.bf.entity.dto.ReconQueueMsg;
|
|
|
import com.hdkj.lt.bf.entity.vo.XlRelVO;
|
|
|
import com.hdkj.lt.bf.event.ReconTriggerEvent;
|
|
|
import com.hdkj.lt.bf.mapper.FhzgSeCurrentEventMapper;
|
|
|
import com.hdkj.lt.bf.mapper.FhzgSeEventStateEstimationMapper;
|
|
|
import com.hdkj.lt.bf.mapper.FhzgSeVoltageEventMapper;
|
|
|
import com.hdkj.lt.bf.mapper.XlRelMapper;
|
|
|
+import com.hdkj.lt.bf.entity.dto.XlRelDTO;
|
|
|
import com.hdkj.lt.core.bizms.modle.dto.StateEstimation;
|
|
|
import com.hdkj.lt.core.bizms.modle.po.DwdShbDsFeederBase;
|
|
|
import com.hdkj.lt.core.bizms.modle.po.Jxz;
|
|
|
@@ -35,22 +35,36 @@ import com.hdkj.lt.modle.vo.recon.ReconstructionParam;
|
|
|
import lombok.RequiredArgsConstructor;
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
import org.springframework.context.event.EventListener;
|
|
|
+import org.springframework.data.redis.connection.stream.Consumer;
|
|
|
+import org.springframework.data.redis.connection.stream.MapRecord;
|
|
|
+import org.springframework.data.redis.connection.stream.ReadOffset;
|
|
|
+import org.springframework.data.redis.connection.stream.RecordId;
|
|
|
+import org.springframework.data.redis.connection.stream.StreamOffset;
|
|
|
+import org.springframework.data.redis.connection.stream.StreamReadOptions;
|
|
|
+import org.springframework.data.redis.connection.stream.StreamRecords;
|
|
|
+import org.springframework.data.redis.core.RedisTemplate;
|
|
|
import org.springframework.scheduling.annotation.Async;
|
|
|
+import org.springframework.scheduling.annotation.Scheduled;
|
|
|
import org.springframework.stereotype.Component;
|
|
|
|
|
|
import java.time.LocalDateTime;
|
|
|
-import java.time.ZoneId;
|
|
|
import java.time.format.DateTimeFormatter;
|
|
|
import java.util.*;
|
|
|
-import java.util.concurrent.*;
|
|
|
+import java.util.concurrent.CompletableFuture;
|
|
|
import java.util.stream.Collectors;
|
|
|
|
|
|
/**
|
|
|
- * 重构算法触发监听器
|
|
|
+ * 重构算法触发监听器（可靠队列版）
|
|
|
* <p>
|
|
|
- * 当线路重过载/电压越限告警持续1小时触发时，异步调用同事编写的重构接口。
|
|
|
+ * 当线路重过载/电压越限告警持续触发时，异步调用同事编写的重构接口。
|
|
|
* 通过 Feign 客户端 IReconApiClient 调用 SI 模块的 POST /recon/trigger。
|
|
|
* <p>
|
|
|
+ * 可靠性改造：原来 @Async + @EventListener 是进程内线程池，服务重启时
|
|
|
+ * 任务直接丢失、reconStatus 卡在"生成中"。现改为 Redis Stream 可靠队列：
|
|
|
+ * <ul>
|
|
|
+ * <li>发布端：事件入队 XADD（Redis Stream，跨进程持久）</li>
|
|
|
+ * <li>消费端：定时轮询 XREADGROUP → 处理后 XACK（重启后 pending 消息不丢）</li>
|
|
|
+ * </ul>
|
|
|
* 调用过程中同步更新告警事件的 reconStatus：
|
|
|
* <ul>
|
|
|
* <li>1 — 重构生成中</li>
|
|
|
@@ -74,6 +88,7 @@ public class ReconTriggerListener {
|
|
|
private final DwdShbDsFeederBaseMapper feederBaseMapper;
|
|
|
private final FhzgSeCurrentEventMapper currentEventMapper;
|
|
|
private final FhzgSeVoltageEventMapper voltageEventMapper;
|
|
|
+ private final RedisTemplate<String, Object> redisTemplate;
|
|
|
private final FhzgSeEventStateEstimationMapper stateEstimationMapper;
|
|
|
private final IStateEstimationApiClient stateEstimationApiClient;
|
|
|
|
|
|
@@ -84,9 +99,201 @@ public class ReconTriggerListener {
|
|
|
private static final int RECON_NO_PLAN = 3;
|
|
|
private static final int RECON_FAILED = -1;
|
|
|
|
|
|
- @Async
|
|
|
+ /** Redis Stream key：重构触发任务队列 */
|
|
|
+ private static final String RECON_STREAM_KEY = "se:recon:queue";
|
|
|
+ /** 消费组名 */
|
|
|
+ private static final String RECON_GROUP = "se-recon-group";
|
|
|
+ /** 消费者名：多副本部署时每实例需唯一，用 HOSTNAME；重启后同名可重新认领自己的 pending 消息 */
|
|
|
+ private final String reconConsumer = "se-recon-consumer-"
|
|
|
+ + (System.getenv("HOSTNAME") != null ? System.getenv("HOSTNAME") : "local");
|
|
|
+
|
|
|
+ // ============================================================
|
|
|
+ // 发布端：事件入队
|
|
|
+ // ============================================================
|
|
|
+
|
|
|
+ /** Redis 连续入队失败计数（熔断用，多线程并发 ++ 需原子） */
|
|
|
+ private final java.util.concurrent.atomic.AtomicInteger redisFailCount = new java.util.concurrent.atomic.AtomicInteger(0);
|
|
|
+ /** Redis 连续失败超过该次数后暂停降级直连，仅记录（防雪崩） */
|
|
|
+ private static final int REDIS_FAIL_THRESHOLD = 5;
|
|
|
+
|
|
|
+ @Async("seAsyncExecutor")
|
|
|
@EventListener
|
|
|
public void onReconTrigger(ReconTriggerEvent event) {
|
|
|
+ try {
|
|
|
+ ReconQueueMsg msg = toQueueMsg(event);
|
|
|
+ Map<String, Object> body = new HashMap<>(2);
|
|
|
+ body.put("payload", JSON.toJSONString(msg));
|
|
|
+ redisTemplate.opsForStream().add(StreamRecords.newRecord()
|
|
|
+ .in(RECON_STREAM_KEY)
|
|
|
+ .ofMap(body));
|
|
|
+ redisFailCount.set(0);
|
|
|
+ log.info("重构触发已入队: eventId={}, feederId={}, alarmType={}, pointTime={}",
|
|
|
+ event.getEventId(), event.getFeederId(), event.getAlarmType(), event.getPointTime());
|
|
|
+ } catch (Exception e) {
|
|
|
+ int failCount = redisFailCount.incrementAndGet();
|
|
|
+ // Redis 不可用时降级：异步直连处理（不阻塞当前线程池）。
|
|
|
+ // 连续失败超阈值后仅记录，不再降级直连（防批量事件触发时全部同步阻塞拖垮线程池）
|
|
|
+ if (failCount <= REDIS_FAIL_THRESHOLD) {
|
|
|
+ log.error("重构触发入队失败，降级直连处理: eventId={}, failCount={}, error={}",
|
|
|
+ event.getEventId(), failCount, e.getMessage(), e);
|
|
|
+ CompletableFuture.runAsync(() -> handleReconTrigger(event));
|
|
|
+ } else {
|
|
|
+ log.error("重构触发入队失败且超过熔断阈值，跳过降级直连(留日志人工处理): eventId={}, failCount={}, error={}",
|
|
|
+ event.getEventId(), failCount, e.getMessage(), e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // ============================================================
|
|
|
+ // 消费端：定时轮询队列并处理（每 10 秒）
|
|
|
+ // ============================================================
|
|
|
+
|
|
|
+ @Scheduled(cron = "0/10 * * * * ?")
|
|
|
+ public void consumeReconQueue() {
|
|
|
+ try {
|
|
|
+ ensureConsumerGroup();
|
|
|
+ List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().read(
|
|
|
+ Consumer.from(RECON_GROUP, reconConsumer),
|
|
|
+ StreamReadOptions.empty().count(20),
|
|
|
+ StreamOffset.create(RECON_STREAM_KEY, ReadOffset.lastConsumed()));
|
|
|
+ if (records == null || records.isEmpty()) return;
|
|
|
+ for (MapRecord<String, Object, Object> record : records) {
|
|
|
+ processQueueRecord(record);
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("重构队列消费异常", e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 处理失败重试上限（超过后丢弃并记录 ERROR，防无限重试） */
|
|
|
+ private static final int MAX_RETRY = 3;
|
|
|
+
|
|
|
+ private void processQueueRecord(MapRecord<String, Object, Object> record) {
|
|
|
+ RecordId id = record.getId();
|
|
|
+ Object payloadObj = null;
|
|
|
+ try {
|
|
|
+ payloadObj = record.getValue().get("payload");
|
|
|
+ if (payloadObj == null) {
|
|
|
+ log.warn("重构队列消息缺少 payload, id={}", id);
|
|
|
+ redisTemplate.opsForStream().acknowledge(RECON_STREAM_KEY, RECON_GROUP, id);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ // FastJson2 序列化 Map 时可能带 autotype 信息，读回形态可能是 String 或 Map，
|
|
|
+ // 统一转成 JSON 字符串再解析
|
|
|
+ String payloadJson;
|
|
|
+ if (payloadObj instanceof Map) {
|
|
|
+ payloadJson = JSON.toJSONString(payloadObj);
|
|
|
+ } else if (payloadObj instanceof byte[]) {
|
|
|
+ payloadJson = new String((byte[]) payloadObj, java.nio.charset.StandardCharsets.UTF_8);
|
|
|
+ } else {
|
|
|
+ payloadJson = String.valueOf(payloadObj);
|
|
|
+ }
|
|
|
+ ReconQueueMsg msg = JSON.parseObject(payloadJson, ReconQueueMsg.class);
|
|
|
+ ReconTriggerEvent event = toEvent(msg);
|
|
|
+ handleReconTrigger(event);
|
|
|
+ // 处理成功才 ACK
|
|
|
+ redisTemplate.opsForStream().acknowledge(RECON_STREAM_KEY, RECON_GROUP, id);
|
|
|
+ } catch (Exception e) {
|
|
|
+ // 失败分类：消息损坏（payload 缺失/JSON 解析失败）→ 重试无意义，直接 ACK 丢弃；
|
|
|
+ // 业务处理失败（handleReconTrigger 抛异常）→ 重新入队重试，超上限丢弃
|
|
|
+ boolean malformed = payloadObj == null || !isParsablePayload(payloadObj);
|
|
|
+ if (malformed) {
|
|
|
+ log.error("重构队列消息损坏(丢弃): id={}, error={}", id, e.getMessage(), e);
|
|
|
+ redisTemplate.opsForStream().acknowledge(RECON_STREAM_KEY, RECON_GROUP, id);
|
|
|
+ } else {
|
|
|
+ handleRetryOrDrop(id, payloadObj, e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 判断 payload 是否能解析（粗略判断，供失败分类用）
|
|
|
+ */
|
|
|
+ private boolean isParsablePayload(Object payloadObj) {
|
|
|
+ try {
|
|
|
+ if (payloadObj instanceof Map) {
|
|
|
+ JSON.toJSONString(payloadObj);
|
|
|
+ } else if (payloadObj instanceof byte[]) {
|
|
|
+ new String((byte[]) payloadObj, java.nio.charset.StandardCharsets.UTF_8);
|
|
|
+ } else {
|
|
|
+ String.valueOf(payloadObj);
|
|
|
+ }
|
|
|
+ return true;
|
|
|
+ } catch (Exception e) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 业务失败：重试次数 < 上限则重新入队（retryCount+1）并 ACK 原消息；
|
|
|
+ * 超过上限则丢弃并记录 ERROR（人工排查）。
|
|
|
+ */
|
|
|
+ private void handleRetryOrDrop(RecordId id, Object payloadObj, Exception cause) {
|
|
|
+ try {
|
|
|
+ String payloadJson;
|
|
|
+ if (payloadObj instanceof Map) {
|
|
|
+ payloadJson = JSON.toJSONString(payloadObj);
|
|
|
+ } else {
|
|
|
+ payloadJson = String.valueOf(payloadObj);
|
|
|
+ }
|
|
|
+ ReconQueueMsg msg = JSON.parseObject(payloadJson, ReconQueueMsg.class);
|
|
|
+ int retry = msg.getRetryCount() == null ? 0 : msg.getRetryCount();
|
|
|
+ if (retry < MAX_RETRY) {
|
|
|
+ msg.setRetryCount(retry + 1);
|
|
|
+ Map<String, Object> body = new HashMap<>(2);
|
|
|
+ body.put("payload", JSON.toJSONString(msg));
|
|
|
+ redisTemplate.opsForStream().add(StreamRecords.newRecord()
|
|
|
+ .in(RECON_STREAM_KEY)
|
|
|
+ .ofMap(body));
|
|
|
+ redisTemplate.opsForStream().acknowledge(RECON_STREAM_KEY, RECON_GROUP, id);
|
|
|
+ log.warn("重构队列消息处理失败，重试入队({}/{}): id={}, error={}",
|
|
|
+ retry + 1, MAX_RETRY, id, cause.getMessage());
|
|
|
+ } else {
|
|
|
+ redisTemplate.opsForStream().acknowledge(RECON_STREAM_KEY, RECON_GROUP, id);
|
|
|
+ log.error("重构队列消息重试超限(丢弃,需人工排查): id={}, retry={}, error={}",
|
|
|
+ id, retry, cause.getMessage(), cause);
|
|
|
+ }
|
|
|
+ } catch (Exception e2) {
|
|
|
+ // 重试入队本身失败：ACK 原消息避免死循环，留日志人工处理
|
|
|
+ log.error("重构队列消息重试入队失败(ACK丢弃): id={}, error={}", id, e2.getMessage(), e2);
|
|
|
+ try {
|
|
|
+ redisTemplate.opsForStream().acknowledge(RECON_STREAM_KEY, RECON_GROUP, id);
|
|
|
+ } catch (Exception ignored) {
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 确保消费组存在（首次启动时创建，MKSTREAM 自动建空流）
|
|
|
+ * <p>
|
|
|
+ * 注意：必须用 ReadOffset.from("0") 从头创建组，不能用默认（$）——
|
|
|
+ * 默认 $ 表示"只接收组创建之后的新消息"，若组创建前已有 XADD 的历史
|
|
|
+ * 消息（如重启期间入队、消费端晚于发布端启动），会被永久跳过。
|
|
|
+ */
|
|
|
+ private void ensureConsumerGroup() {
|
|
|
+ try {
|
|
|
+ redisTemplate.opsForStream().createGroup(RECON_STREAM_KEY, ReadOffset.from("0"), RECON_GROUP);
|
|
|
+ } catch (Exception e) {
|
|
|
+ // BUSYGROUP 已存在则忽略；其余异常抛出
|
|
|
+ String msg = e.getMessage() == null ? "" : e.getMessage();
|
|
|
+ if (!msg.contains("BUSYGROUP")) {
|
|
|
+ throw e;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // ============================================================
|
|
|
+ // 核心处理逻辑（原 onReconTrigger 主体）
|
|
|
+ // ============================================================
|
|
|
+
|
|
|
+ private void handleReconTrigger(ReconTriggerEvent event) {
|
|
|
+ // 幂等保护：事件已完成重构（有方案/无方案）则跳过，防重投/重复触发重复调用重构接口
|
|
|
+ Integer doneStatus = queryReconStatus(event.getEventId(), event.getAlarmType());
|
|
|
+ if (doneStatus != null && (doneStatus == RECON_COMPLETED || doneStatus == RECON_NO_PLAN)) {
|
|
|
+ log.info("重构已处理过(跳过): eventId={}, reconStatus={}", event.getEventId(), doneStatus);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
updateReconStatus(event.getEventId(), event.getAlarmType(), RECON_RUNNING);
|
|
|
|
|
|
Map<String, Object> params = new HashMap<>();
|
|
|
@@ -200,22 +407,51 @@ public class ReconTriggerListener {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * 解析重构接口响应（真实形状: ApiResponse{code, success, data:{success,...}, msg}）
|
|
|
+ * <p>
|
|
|
+ * code=10000(ResultCode.SUCCESS) 且 data.success=true → 有方案(2)
|
|
|
+ * code=10000 且 data.success=false → 无方案(3)
|
|
|
+ * code=10001(业务失败) / 解析失败 → 调用失败(-1)
|
|
|
+ */
|
|
|
private int parseReconResponse(String response) {
|
|
|
if (response == null || response.isEmpty()) return RECON_FAILED;
|
|
|
try {
|
|
|
JSONObject root = com.alibaba.fastjson.JSON.parseObject(response);
|
|
|
- String status = root.getString("status");
|
|
|
- if ("200".equals(status) || "success".equalsIgnoreCase(status)) {
|
|
|
+ int code = root.getIntValue("code");
|
|
|
+ if (code == 10000) {
|
|
|
Object data = root.get("data");
|
|
|
if (data instanceof JSONObject) {
|
|
|
- Boolean success = ((JSONObject) data).getBoolean("success");
|
|
|
- return Boolean.TRUE.equals(success) ? RECON_COMPLETED : RECON_NO_PLAN;
|
|
|
+ Boolean dataSuccess = ((JSONObject) data).getBoolean("success");
|
|
|
+ return Boolean.TRUE.equals(dataSuccess) ? RECON_COMPLETED : RECON_NO_PLAN;
|
|
|
}
|
|
|
+ // data 缺失/非对象：按成功处理（接口正常返回但无明细，视为有方案）
|
|
|
+ return RECON_COMPLETED;
|
|
|
}
|
|
|
+ return RECON_FAILED;
|
|
|
} catch (Exception e) {
|
|
|
log.warn("解析重构接口响应失败: {}", e.getMessage());
|
|
|
+ return RECON_FAILED;
|
|
|
}
|
|
|
- return RECON_COMPLETED;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 查询事件当前 reconStatus（null = 事件不存在或未处理过）
|
|
|
+ */
|
|
|
+ private Integer queryReconStatus(Long eventId, String alarmType) {
|
|
|
+ if (eventId == null) return null;
|
|
|
+ try {
|
|
|
+ if (alarmType != null && alarmType.startsWith("current_")) {
|
|
|
+ FhzgSeCurrentEvent cur = currentEventMapper.selectById(eventId);
|
|
|
+ return cur != null ? cur.getReconStatus() : null;
|
|
|
+ } else if (alarmType != null && alarmType.startsWith("voltage_")) {
|
|
|
+ FhzgSeVoltageEvent cur = voltageEventMapper.selectById(eventId);
|
|
|
+ return cur != null ? cur.getReconStatus() : null;
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.warn("查询事件 reconStatus 失败: eventId={}, error={}", eventId, e.getMessage());
|
|
|
+ }
|
|
|
+ return null;
|
|
|
}
|
|
|
|
|
|
private void updateReconStatus(Long eventId, String alarmType, int reconStatus) {
|
|
|
@@ -238,6 +474,33 @@ public class ReconTriggerListener {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ // ============================================================
|
|
|
+ // 消息转换
|
|
|
+ // ============================================================
|
|
|
+
|
|
|
+ private ReconQueueMsg toQueueMsg(ReconTriggerEvent e) {
|
|
|
+ ReconQueueMsg msg = new ReconQueueMsg();
|
|
|
+ msg.setEventId(e.getEventId());
|
|
|
+ msg.setFeederId(e.getFeederId());
|
|
|
+ msg.setFeederName(e.getFeederName());
|
|
|
+ msg.setCountyId(e.getCountyId());
|
|
|
+ msg.setSubsId(e.getSubsId());
|
|
|
+ msg.setAlarmType(e.getAlarmType());
|
|
|
+ msg.setPointTime(e.getPointTime() != null ? e.getPointTime().format(DT_FMT) : null);
|
|
|
+ msg.setTriggerTime(e.getTriggerTime() != null ? e.getTriggerTime().format(DT_FMT) : null);
|
|
|
+ return msg;
|
|
|
+ }
|
|
|
+
|
|
|
+ private ReconTriggerEvent toEvent(ReconQueueMsg msg) {
|
|
|
+ LocalDateTime pointTime = msg.getPointTime() != null
|
|
|
+ ? LocalDateTime.parse(msg.getPointTime(), DT_FMT) : null;
|
|
|
+ LocalDateTime triggerTime = msg.getTriggerTime() != null
|
|
|
+ ? LocalDateTime.parse(msg.getTriggerTime(), DT_FMT) : null;
|
|
|
+ return new ReconTriggerEvent(
|
|
|
+ msg.getEventId(), msg.getFeederId(), msg.getFeederName(),
|
|
|
+ msg.getCountyId(), msg.getSubsId(), msg.getAlarmType(),
|
|
|
+ pointTime, triggerTime);
|
|
|
+ }
|
|
|
/**
|
|
|
* 状估数据保存
|
|
|
* @param event
|
liuaini