wuweihua/BMS
2
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增游标抓取逻辑，优化UUID到新同步服务的查询和Redis配置

Browse Source
This commit is contained in:
wuweihua
2026-11-15 12:26:12 +08:00
parent 206e4e786b
commit 63ded27fd5
7 changed files with 192 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bms-back/bms-system/src/main/java/com/bms/price/migration/domain/UuidToNewPendingRow.java
+17
View File
@@ -0,0 +1,17 @@
package com.bms.price.migration.domain;
import java.util.Date;
/**
* 待同步订单（含游标 create_time）
*/
public class UuidToNewPendingRow
{
private String billId;
private Date createTime;
public String getBillId() { return billId; }
public void setBillId(String billId) { this.billId = billId; }
public Date getCreateTime() { return createTime; }
public void setCreateTime(Date createTime) { this.createTime = createTime; }
}
bms-back/bms-system/src/main/java/com/bms/price/migration/mapper/UuidToNewSyncMapper.java
+13 -1
View File
@@ -4,6 +4,8 @@ import java.util.List;
import org.apache.ibatis.annotations.Param;
import com.bms.price.migration.domain.MigBiJobLog;
import com.bms.price.migration.domain.UuidToNewFeeRow;
import com.bms.price.migration.domain.UuidToNewPendingRow;
import java.util.Date;
import com.bms.price.migration.domain.UuidToNewIdHolder;
import com.bms.price.migration.domain.UuidToNewProgressVo;
@@ -12,7 +14,17 @@ import com.bms.price.migration.domain.UuidToNewProgressVo;
*/
public interface UuidToNewSyncMapper
{
List<String> selectPendingBillIds(@Param("limit") int limit);
/**
* 游标抓取未处理订单（避免每批从最早订单全量排序）
* @param lastCreateTime 上批最后 create_time；空=从头
* @param lastBillId 上批最后 bill_id；与 time 配对
*/
List<UuidToNewPendingRow> selectPendingBillIds(@Param("limit") int limit,
@Param("lastCreateTime") Date lastCreateTime,
@Param("lastBillId") String lastBillId);
/** done 已有进度且 Redis 无游标时：用最新 done 边界初始化 */
UuidToNewPendingRow selectPendingCursorBootstrap();
String selectOrderNo(@Param("oldBillId") String oldBillId);
bms-back/bms-system/src/main/java/com/bms/price/migration/service/impl/UuidToNewSyncServiceImpl.java
+121 -2
View File
@@ -1,7 +1,10 @@
package com.bms.price.migration.service.impl;
import java.nio.charset.StandardCharsets;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -19,8 +22,10 @@ import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import com.bms.common.core.redis.RedisCache;
import com.bms.price.config.UuidToNewProperties;
import com.bms.price.migration.domain.UuidToNewBatchResult;
import com.bms.price.migration.domain.UuidToNewPendingRow;
import com.bms.price.migration.domain.UuidToNewFeeRow;
import com.bms.price.migration.domain.UuidToNewIdHolder;
import com.bms.price.migration.mapper.UuidToNewSyncMapper;
@@ -28,7 +33,7 @@ import com.bms.price.migration.service.IUuidToNewSyncService;
/**
* UUID -> bigint 阴影同步：订单+income *_new + 关联表回写 order_id。
* 占坑独立提交；一单一事务；批内可多线程。
* 占坑独立提交；一单一事务；批内可多线程；游标抓取避全表排序。
*/
@Service
public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
@@ -46,10 +51,19 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
@Autowired
private UuidToNewProperties properties;
@Autowired
private RedisCache redisCache;
@Autowired
@Lazy
private IUuidToNewSyncService self;
/** Redis：抓取游标 create_time */
public static final String REDIS_KEY_CURSOR_TIME = "bms:mig:uuid-to-new:cursor-time";
/** Redis：抓取游标 bill_id */
public static final String REDIS_KEY_CURSOR_BILL_ID = "bms:mig:uuid-to-new:cursor-bill-id";
private static final String CURSOR_TIME_PATTERN = "yyyy-MM-dd HH:mm:ss";
@Override
public UuidToNewBatchResult syncBatch(int batchSize)
{
@@ -60,7 +74,30 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
job.setBatchNo(batchNo);
syncMapper.insertJobLogStart(job);
List<String> billIds = syncMapper.selectPendingBillIds(limit);
Date lastCreateTime = loadCursorTime();
String lastBillId = loadCursorBillId();
if (lastCreateTime == null || lastBillId == null || lastBillId.isEmpty())
{
// done 已有进度时用边界初始化，避免从最早订单重扫
UuidToNewPendingRow boot = syncMapper.selectPendingCursorBootstrap();
if (boot != null && boot.getCreateTime() != null && boot.getBillId() != null)
{
lastCreateTime = boot.getCreateTime();
lastBillId = boot.getBillId();
saveCursor(lastCreateTime, lastBillId);
log.info("uuid-to-new cursor bootstrap billId={} createTime={}", lastBillId, lastCreateTime);
}
}
List<UuidToNewPendingRow> pending = syncMapper.selectPendingBillIds(limit, lastCreateTime, lastBillId);
List<String> billIds = new ArrayList<String>();
for (UuidToNewPendingRow row : pending)
{
if (row != null && row.getBillId() != null)
{
billIds.add(row.getBillId());
}
}
AtomicInteger success = new AtomicInteger(0);
AtomicInteger fail = new AtomicInteger(0);
AtomicInteger skip = new AtomicInteger(0);
@@ -139,6 +176,16 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
job.setFailReason(abbreviate(remark, 480));
syncMapper.updateJobLogFinish(job);
// 批次结束后推进游标（本批已尝试的最后一单）
if (!pending.isEmpty())
{
UuidToNewPendingRow last = pending.get(pending.size() - 1);
if (last.getCreateTime() != null && last.getBillId() != null)
{
saveCursor(last.getCreateTime(), last.getBillId());
}
}
UuidToNewBatchResult result = new UuidToNewBatchResult();
result.setFetched(billIds.size());
result.setSuccess(success.get());
@@ -371,4 +418,76 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
syncMapper.updateBillingTaskOrderId(oldBillId, newBillId);
syncMapper.updateArGenTaskOrderId(oldBillId, newBillId);
}
private Date loadCursorTime()
{
if (redisCache == null)
{
return null;
}
try
{
Object v = redisCache.getCacheObject(REDIS_KEY_CURSOR_TIME);
if (v == null)
{
return null;
}
if (v instanceof Date)
{
return (Date) v;
}
String s = String.valueOf(v).trim();
if (s.isEmpty())
{
return null;
}
return new SimpleDateFormat(CURSOR_TIME_PATTERN).parse(s);
}
catch (ParseException pe)
{
log.warn("uuid-to-new parse cursor time failed: {}", pe.toString());
return null;
}
catch (Exception e)
{
log.warn("uuid-to-new load cursor time failed: {}", e.toString());
return null;
}
}
private String loadCursorBillId()
{
if (redisCache == null)
{
return null;
}
try
{
Object v = redisCache.getCacheObject(REDIS_KEY_CURSOR_BILL_ID);
return v == null ? null : String.valueOf(v).trim();
}
catch (Exception e)
{
log.warn("uuid-to-new load cursor billId failed: {}", e.toString());
return null;
}
}
private void saveCursor(Date createTime, String billId)
{
if (redisCache == null || createTime == null || billId == null || billId.isEmpty())
{
return;
}
try
{
String t = new SimpleDateFormat(CURSOR_TIME_PATTERN).format(createTime);
redisCache.setCacheObject(REDIS_KEY_CURSOR_TIME, t);
redisCache.setCacheObject(REDIS_KEY_CURSOR_BILL_ID, billId);
}
catch (Exception e)
{
log.warn("uuid-to-new save cursor failed: {}", e.toString());
}
}
}
bms-back/bms-system/src/main/java/com/bms/price/task/UuidToNewSyncRunner.java
+1
View File
@@ -18,6 +18,7 @@ import com.bms.price.migration.service.IUuidToNewSyncService;
* UUID 阴影同步运行器：监控页开关控制；
* 开启后跑批；全失败会退避并自动关闭，避免把库打满。
* 开关/线程/批次落 Redis，服务重启后按 Redis 状态恢复。
* 抓取游标由 SyncService 维护（bms:mig:uuid-to-new:cursor-time / cursor-bill-id）。
*/
@Component
@Order(100)
bms-back/bms-system/src/main/resources/mapper/price/UuidToNewSyncMapper.xml
+21 -4
View File
@@ -4,15 +4,32 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.bms.price.migration.mapper.UuidToNewSyncMapper">
<select id="selectPendingBillIds" resultType="String">
select o.bill_id
<select id="selectPendingBillIds" resultType="com.bms.price.migration.domain.UuidToNewPendingRow">
select o.bill_id as billId, o.create_time as createTime
from prc_order_bill o
left join mig_bi_order_done d on d.old_bill_id = o.bill_id
where d.old_bill_id is null
where 1 = 1
<if test="lastCreateTime != null and lastBillId != null and lastBillId != ''">
and (
o.create_time &gt; #{lastCreateTime}
or (o.create_time = #{lastCreateTime} and o.bill_id &gt; #{lastBillId})
)
</if>
and not exists (
select 1 from mig_bi_order_done d
where d.old_bill_id = o.bill_id
)
order by o.create_time, o.bill_id
limit #{limit}
</select>
<select id="selectPendingCursorBootstrap" resultType="com.bms.price.migration.domain.UuidToNewPendingRow">
select o.bill_id as billId, o.create_time as createTime
from mig_bi_order_done d
inner join prc_order_bill o on o.bill_id = d.old_bill_id
order by o.create_time desc, o.bill_id desc
limit 1
</select>
<select id="selectOrderNo" resultType="String">
select order_no from prc_order_bill where bill_id = #{oldBillId} limit 1
</select>