autumn/jsowell-charger-web
1
0
Fork 0
mirror of https://codeup.aliyun.com/67c68d4e484ca2f0a13ac3c1/ydc/jsowell-charger-web.git synced 2026-07-14 19:18:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

优化pile_msg_record插入数据

Browse Source
This commit is contained in:
jsowell
2026-07-01 16:30:38 +08:00
parent 06aab00627
commit aa7a9a10d9
5 changed files with 153 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
docs/sql/pile_msg_record_retention_index.sql Normal file
View File

@@ -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`);

9
jsowell-pile/src/main/java/com/jsowell/pile/mapper/PileMsgRecordMapper.java
View File

@@ -23,4 +23,13 @@ public interface PileMsgRecordMapper {
List<PileMsgRecord> queryPileFeedList(@Param("pileSn") String pileSn, @Param("frameType") String frameType,
@Param("transactionCode") String transactionCode);
List<PileMsgRecord> 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);
}

99
jsowell-pile/src/main/java/com/jsowell/pile/service/impl/PileMsgRecordRetentionService.java Normal file
View File

@@ -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<String> 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;
}
}

9
jsowell-pile/src/main/java/com/jsowell/pile/service/impl/PileMsgRecordServiceImpl.java
View File

@@ -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<PileMsgRecord> queryPileFeedList(String pileSn, String frameType, String transactionCode) {
List<PileMsgRecord> 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;
}

34
jsowell-pile/src/main/resources/mapper/pile/PileMsgRecordMapper.xml
View File

@@ -16,7 +16,7 @@
</resultMap>
<sql id="Base_Column_List">
<!--@mbg.generated-->
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
</sql>
<select id="selectByPrimaryKey" parameterType="java.lang.Integer" resultMap="BaseResultMap">
<!--@mbg.generated-->
@@ -154,6 +154,19 @@
</select>
<select id="queryPileFeedList" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from pile_msg_record
where pile_sn = #{pileSn,jdbcType=VARCHAR}
<if test="frameType != null and frameType != ''">
and frame_type = #{frameType,jdbcType=VARCHAR}
</if>
<if test="transactionCode != null and transactionCode != ''" >
and transaction_code = #{transactionCode,jdbcType=VARCHAR}
</if>
</select>
<select id="queryPileFeedListByJsonTransactionCode" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from pile_msg_record
@@ -165,4 +178,23 @@
and json_msg->>'$.transactionCode' LIKE CONCAT('%', #{transactionCode}, '%')
</if>
</select>
<select id="selectRetentionCutoffId" resultType="java.lang.Long">
select id
from pile_msg_record
where pile_sn = #{pileSn,jdbcType=VARCHAR}
and frame_type = #{frameType,jdbcType=VARCHAR}
and frame_type != '0x3B'
order by id desc
limit #{offset}, 1
</select>
<delete id="deleteOldRecordsBeforeId">
delete from pile_msg_record
where pile_sn = #{pileSn,jdbcType=VARCHAR}
and frame_type = #{frameType,jdbcType=VARCHAR}
and frame_type != '0x3B'
and id &lt; #{cutoffId,jdbcType=BIGINT}
limit #{limit}
</delete>
</mapper>