diff --git a/docs/sql/pile_msg_record_retention_index.sql b/docs/sql/pile_msg_record_retention_index.sql new file mode 100644 index 000000000..11a0b2fbe --- /dev/null +++ b/docs/sql/pile_msg_record_retention_index.sql @@ -0,0 +1,3 @@ +ALTER TABLE `pile_msg_record` +ADD KEY `idx_pile_frame_id` (`pile_sn`, `frame_type`, `id`), +ADD KEY `idx_pile_frame_transaction` (`pile_sn`, `frame_type`, `transaction_code`); diff --git a/jsowell-pile/src/main/java/com/jsowell/pile/mapper/PileMsgRecordMapper.java b/jsowell-pile/src/main/java/com/jsowell/pile/mapper/PileMsgRecordMapper.java index 8e2779937..447a37d7f 100644 --- a/jsowell-pile/src/main/java/com/jsowell/pile/mapper/PileMsgRecordMapper.java +++ b/jsowell-pile/src/main/java/com/jsowell/pile/mapper/PileMsgRecordMapper.java @@ -23,4 +23,13 @@ public interface PileMsgRecordMapper { List queryPileFeedList(@Param("pileSn") String pileSn, @Param("frameType") String frameType, @Param("transactionCode") String transactionCode); + + List queryPileFeedListByJsonTransactionCode(@Param("pileSn") String pileSn, @Param("frameType") String frameType, + @Param("transactionCode") String transactionCode); + + Long selectRetentionCutoffId(@Param("pileSn") String pileSn, @Param("frameType") String frameType, + @Param("offset") int offset); + + int deleteOldRecordsBeforeId(@Param("pileSn") String pileSn, @Param("frameType") String frameType, + @Param("cutoffId") Long cutoffId, @Param("limit") int limit); } diff --git a/jsowell-pile/src/main/java/com/jsowell/pile/service/impl/PileMsgRecordRetentionService.java b/jsowell-pile/src/main/java/com/jsowell/pile/service/impl/PileMsgRecordRetentionService.java new file mode 100644 index 000000000..f72db6faf --- /dev/null +++ b/jsowell-pile/src/main/java/com/jsowell/pile/service/impl/PileMsgRecordRetentionService.java @@ -0,0 +1,99 @@ +package com.jsowell.pile.service.impl; + +import com.jsowell.common.util.StringUtils; +import com.jsowell.pile.mapper.PileMsgRecordMapper; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Service; + +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * 非交易记录报文保留策略: + * 每个桩的每个帧类型仅保留最近200条，删除动作放到后台异步分批执行， + * 避免在报文写入主链路上同步删除带来额外延迟。 + */ +@Slf4j +@Service +public class PileMsgRecordRetentionService { + + private static final String KEEP_FOREVER_FRAME_TYPE = "0x3B"; + private static final int RETAIN_COUNT = 200; + private static final int RETENTION_OFFSET = RETAIN_COUNT - 1; + private static final int DELETE_BATCH_SIZE = 500; + private static final int MAX_DELETE_BATCHES_PER_RUN = 20; + private static final long CLEANUP_DELAY_SECONDS = 30L; + private static final long FOLLOW_UP_DELAY_SECONDS = 5L; + + @Autowired + private PileMsgRecordMapper pileMsgRecordMapper; + + @Autowired + @Qualifier("scheduledExecutorService") + private ScheduledExecutorService scheduledExecutorService; + + private final Set pendingCleanupKeys = ConcurrentHashMap.newKeySet(); + + public void scheduleRetention(String pileSn, String frameType) { + if (StringUtils.isBlank(pileSn) || StringUtils.isBlank(frameType) || KEEP_FOREVER_FRAME_TYPE.equals(frameType)) { + return; + } + this.scheduleRetention(pileSn, frameType, CLEANUP_DELAY_SECONDS); + } + + private void scheduleRetention(String pileSn, String frameType, long delaySeconds) { + String key = buildKey(pileSn, frameType); + if (!pendingCleanupKeys.add(key)) { + return; + } + scheduledExecutorService.schedule(() -> runCleanup(key, pileSn, frameType), delaySeconds, TimeUnit.SECONDS); + } + + private void runCleanup(String key, String pileSn, String frameType) { + boolean hasMore = false; + try { + hasMore = cleanupGroup(pileSn, frameType); + } catch (Exception e) { + log.error("pile_msg_record清理失败, pileSn:{}, frameType:{}", pileSn, frameType, e); + } finally { + pendingCleanupKeys.remove(key); + } + + if (hasMore) { + this.scheduleRetention(pileSn, frameType, FOLLOW_UP_DELAY_SECONDS); + } + } + + private boolean cleanupGroup(String pileSn, String frameType) { + Long cutoffId = pileMsgRecordMapper.selectRetentionCutoffId(pileSn, frameType, RETENTION_OFFSET); + if (cutoffId == null) { + return false; + } + + int totalDeleted = 0; + for (int i = 0; i < MAX_DELETE_BATCHES_PER_RUN; i++) { + int deleted = pileMsgRecordMapper.deleteOldRecordsBeforeId(pileSn, frameType, cutoffId, DELETE_BATCH_SIZE); + totalDeleted += deleted; + if (deleted < DELETE_BATCH_SIZE) { + if (totalDeleted > 0) { + log.info("pile_msg_record清理完成, pileSn:{}, frameType:{}, deleted:{}", pileSn, frameType, totalDeleted); + } + return false; + } + } + + if (totalDeleted > 0) { + log.info("pile_msg_record本轮清理完成, pileSn:{}, frameType:{}, deleted:{}, 将继续异步清理", + pileSn, frameType, totalDeleted); + } + return true; + } + + private String buildKey(String pileSn, String frameType) { + return pileSn + "#" + frameType; + } +} diff --git a/jsowell-pile/src/main/java/com/jsowell/pile/service/impl/PileMsgRecordServiceImpl.java b/jsowell-pile/src/main/java/com/jsowell/pile/service/impl/PileMsgRecordServiceImpl.java index 1b306bbc0..cd0548469 100644 --- a/jsowell-pile/src/main/java/com/jsowell/pile/service/impl/PileMsgRecordServiceImpl.java +++ b/jsowell-pile/src/main/java/com/jsowell/pile/service/impl/PileMsgRecordServiceImpl.java @@ -18,6 +18,7 @@ import com.jsowell.pile.service.OrderBasicInfoService; import com.jsowell.pile.service.PileMsgRecordService; import com.jsowell.pile.vo.web.PileCommunicationLogVO; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.collections4.CollectionUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -36,6 +37,9 @@ public class PileMsgRecordServiceImpl implements PileMsgRecordService { @Autowired private OrderBasicInfoService orderBasicInfoService; + @Autowired + private PileMsgRecordRetentionService pileMsgRecordRetentionService; + @Override public void save(String pileSn, String connectorCode, String frameType, String jsonMsg, String originalMsg) { PileMsgRecord pileMsgRecord = PileMsgRecord.builder() @@ -46,6 +50,7 @@ public class PileMsgRecordServiceImpl implements PileMsgRecordService { .originalMsg(originalMsg) .build(); pileMsgRecordMapper.insertSelective(pileMsgRecord); + pileMsgRecordRetentionService.scheduleRetention(pileSn, frameType); } @Override @@ -85,6 +90,7 @@ public class PileMsgRecordServiceImpl implements PileMsgRecordService { pileMsgRecord.setOriginalMsg(originalMsg); pileMsgRecord.setCreateTime(DateUtils.getDateTime()); pileMsgRecordMapper.insertSelective(pileMsgRecord); + pileMsgRecordRetentionService.scheduleRetention(pileSn, frameType); } // @Override @@ -178,6 +184,9 @@ public class PileMsgRecordServiceImpl implements PileMsgRecordService { @Override public List queryPileFeedList(String pileSn, String frameType, String transactionCode) { List list = pileMsgRecordMapper.queryPileFeedList(pileSn, frameType, transactionCode); + if (CollectionUtils.isEmpty(list) && StringUtils.isNotBlank(transactionCode)) { + list = pileMsgRecordMapper.queryPileFeedListByJsonTransactionCode(pileSn, frameType, transactionCode); + } log.info("queryPileFeedList - pileSn:{}, frameType:{}, transactionCode:{}, list: {}", pileSn, frameType, transactionCode, list); return list; } diff --git a/jsowell-pile/src/main/resources/mapper/pile/PileMsgRecordMapper.xml b/jsowell-pile/src/main/resources/mapper/pile/PileMsgRecordMapper.xml index a3e10b05b..106bb4809 100644 --- a/jsowell-pile/src/main/resources/mapper/pile/PileMsgRecordMapper.xml +++ b/jsowell-pile/src/main/resources/mapper/pile/PileMsgRecordMapper.xml @@ -16,7 +16,7 @@ - id, pile_sn, frame_type, connector_code, pile_connector_code, original_msg, json_msg, create_time + id, pile_sn, transaction_code, frame_type, connector_code, pile_connector_code, original_msg, json_msg, create_time + + + + + + + delete from pile_msg_record + where pile_sn = #{pileSn,jdbcType=VARCHAR} + and frame_type = #{frameType,jdbcType=VARCHAR} + and frame_type != '0x3B' + and id < #{cutoffId,jdbcType=BIGINT} + limit #{limit} +