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

新增UUID到新同步服务的失败处理逻辑和批次结果返回

Browse Source
This commit is contained in:
wuweihua
2026-11-14 19:52:38 +08:00
parent ee36c8c902
commit 358e5a46b9
7 changed files with 223 additions and 32 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/UuidToNewBatchResult.java
+25
View File
@@ -0,0 +1,25 @@
package com.bms.price.migration.domain;
/**
* 一批阴影同步结果（供运行器做失败退避/自动停开）
*/
public class UuidToNewBatchResult
{
private int fetched;
private int success;
private int fail;
private int skip;
/** 本批首条失败原因（截断） */
private String sampleError;
public int getFetched() { return fetched; }
public void setFetched(int fetched) { this.fetched = fetched; }
public int getSuccess() { return success; }
public void setSuccess(int success) { this.success = success; }
public int getFail() { return fail; }
public void setFail(int fail) { this.fail = fail; }
public int getSkip() { return skip; }
public void setSkip(int skip) { this.skip = skip; }
public String getSampleError() { return sampleError; }
public void setSampleError(String sampleError) { this.sampleError = sampleError; }
}
bms-back/bms-system/src/main/java/com/bms/price/migration/mapper/UuidToNewSyncMapper.java
+2
View File
@@ -27,6 +27,8 @@ public interface UuidToNewSyncMapper
int updateOrderDoneSuccess(UuidToNewIdHolder holder);
int updateOrderDoneFail(UuidToNewIdHolder holder);
int insertOrderExtNew(@Param("oldBillId") String oldBillId, @Param("newBillId") Long newBillId);
List<String> selectIncomeIds(@Param("oldBillId") String oldBillId);
bms-back/bms-system/src/main/java/com/bms/price/migration/service/IUuidToNewSyncService.java
+10 -2
View File
@@ -1,5 +1,7 @@
package com.bms.price.migration.service;
import com.bms.price.migration.domain.UuidToNewBatchResult;
/**
* UUID -> bigint 阴影同步服务
*/
@@ -7,10 +9,16 @@ public interface IUuidToNewSyncService
{
/**
* 执行一批。
* @return 本批抓取条数（fetched）
* @return 本批统计（抓取/成功/失败/跳过）
*/
int syncBatch(int batchSize);
UuidToNewBatchResult syncBatch(int batchSize);
/** 一单一事务；返回 SUCCESS / FAIL / SKIP */
String syncOneOrder(String oldBillId);
/** 独立事务占坑；false=已被占用 */
boolean tryClaimOrder(String oldBillId);
/** 独立事务标记 FAIL，避免失败回滚后被无限重试 */
void markOrderFail(String oldBillId, String failReason);
}
bms-back/bms-system/src/main/java/com/bms/price/migration/service/impl/UuidToNewSyncServiceImpl.java
+116 -23
View File
@@ -11,20 +11,23 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
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.price.config.UuidToNewProperties;
import com.bms.price.migration.domain.UuidToNewBatchResult;
import com.bms.price.migration.domain.UuidToNewFeeRow;
import com.bms.price.migration.domain.UuidToNewIdHolder;
import com.bms.price.migration.mapper.UuidToNewSyncMapper;
import com.bms.price.migration.service.IUuidToNewSyncService;
/**
* UUID -> bigint 阴影同步：一单一事务；批内可多线程
* UUID -> bigint 阴影同步：占坑独立提交；业务拷贝一单一事务；批内可多线程
*/
@Service
public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
@@ -47,7 +50,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
private IUuidToNewSyncService self;
@Override
public int syncBatch(int batchSize)
public UuidToNewBatchResult syncBatch(int batchSize)
{
int limit = Math.max(1, Math.min(batchSize, 1000));
int threads = properties == null ? 1 : properties.resolvedThreadCount();
@@ -60,6 +63,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
AtomicInteger success = new AtomicInteger(0);
AtomicInteger fail = new AtomicInteger(0);
AtomicInteger skip = new AtomicInteger(0);
AtomicReference<String> sampleError = new AtomicReference<String>();
if (billIds.isEmpty())
{
@@ -69,7 +73,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
{
for (String oldBillId : billIds)
{
applyResult(self, oldBillId, success, fail, skip);
applyResult(self, oldBillId, success, fail, skip, sampleError);
}
}
else
@@ -86,7 +90,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
@Override
public void run()
{
applyResult(self, id, success, fail, skip);
applyResult(self, id, success, fail, skip, sampleError);
}
}));
}
@@ -99,6 +103,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
catch (Exception e)
{
fail.incrementAndGet();
sampleError.compareAndSet(null, abbreviate(e));
log.warn("uuid-to-new worker future error: {}", e.toString());
}
}
@@ -125,17 +130,29 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
job.setSuccess(success.get());
job.setFail(fail.get());
job.setSkip(skip.get());
job.setFailReason("threads=" + threads);
String remark = "threads=" + threads;
if (sampleError.get() != null)
{
remark = remark + "; err=" + sampleError.get();
}
job.setFailReason(abbreviate(remark, 480));
syncMapper.updateJobLogFinish(job);
String summary = String.format(
"batch=%s threads=%d fetched=%d success=%d fail=%d skip=%d",
batchNo, threads, billIds.size(), success.get(), fail.get(), skip.get());
log.info("uuid-to-new {}", summary);
return billIds.size();
UuidToNewBatchResult result = new UuidToNewBatchResult();
result.setFetched(billIds.size());
result.setSuccess(success.get());
result.setFail(fail.get());
result.setSkip(skip.get());
result.setSampleError(sampleError.get());
log.info("uuid-to-new batch={} threads={} fetched={} success={} fail={} skip={} err={}",
batchNo, threads, result.getFetched(), result.getSuccess(), result.getFail(),
result.getSkip(), result.getSampleError());
return result;
}
private void applyResult(IUuidToNewSyncService svc, String oldBillId,
AtomicInteger success, AtomicInteger fail, AtomicInteger skip)
AtomicInteger success, AtomicInteger fail, AtomicInteger skip,
AtomicReference<String> sampleError)
{
try
{
@@ -148,6 +165,10 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
{
skip.incrementAndGet();
}
else if (FAIL.equals(r))
{
fail.incrementAndGet();
}
else
{
fail.incrementAndGet();
@@ -156,6 +177,16 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
catch (Exception e)
{
fail.incrementAndGet();
String msg = abbreviate(e);
sampleError.compareAndSet(null, msg);
try
{
svc.markOrderFail(oldBillId, msg);
}
catch (Exception markEx)
{
log.warn("uuid-to-new mark FAIL failed billId={}: {}", oldBillId, markEx.toString());
}
log.warn("uuid-to-new sync fail billId={}: {}", oldBillId, e.toString());
}
}
@@ -177,6 +208,78 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
return h;
}
private static String abbreviate(Throwable e)
{
if (e == null)
{
return null;
}
Throwable root = e;
while (root.getCause() != null && root.getCause() != root)
{
root = root.getCause();
}
String msg = root.getClass().getSimpleName() + ": " + root.getMessage();
return abbreviate(msg, 450);
}
private static String abbreviate(String s, int max)
{
if (s == null)
{
return null;
}
String t = s.replace('\r', ' ').replace('\n', ' ');
if (t.length() <= max)
{
return t;
}
return t.substring(0, max);
}
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
public boolean tryClaimOrder(String oldBillId)
{
if (oldBillId == null || oldBillId.isEmpty())
{
return false;
}
if (syncMapper.countOrderBill(oldBillId) <= 0)
{
return false;
}
String orderNo = syncMapper.selectOrderNo(oldBillId);
UuidToNewIdHolder claim = new UuidToNewIdHolder();
claim.setOldBillId(oldBillId);
claim.setNewBillId(claimTempNewBillId(oldBillId));
claim.setOrderNo(orderNo);
claim.setStatus(RUNNING);
return syncMapper.claimOrderDone(claim) > 0;
}
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
public void markOrderFail(String oldBillId, String failReason)
{
if (oldBillId == null || oldBillId.isEmpty())
{
return;
}
UuidToNewIdHolder h = new UuidToNewIdHolder();
h.setOldBillId(oldBillId);
h.setFailReason(abbreviate(failReason, 480));
int n = syncMapper.updateOrderDoneFail(h);
if (n <= 0)
{
// 占坑未成功时补一条 FAIL，阻止再被抓取
h.setNewBillId(claimTempNewBillId(oldBillId));
h.setOrderNo(syncMapper.selectOrderNo(oldBillId));
h.setStatus(FAIL);
syncMapper.insertOrderDone(h);
}
}
@Override
@Transactional(rollbackFor = Exception.class)
public String syncOneOrder(String oldBillId)
@@ -185,18 +288,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
{
return SKIP;
}
if (syncMapper.countOrderBill(oldBillId) <= 0)
{
return SKIP;
}
String orderNo = syncMapper.selectOrderNo(oldBillId);
UuidToNewIdHolder claim = new UuidToNewIdHolder();
claim.setOldBillId(oldBillId);
claim.setNewBillId(claimTempNewBillId(oldBillId));
claim.setOrderNo(orderNo);
claim.setStatus(RUNNING);
if (syncMapper.claimOrderDone(claim) == 0)
if (!self.tryClaimOrder(oldBillId))
{
return SKIP;
}
@@ -215,6 +307,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
throw new IllegalStateException("insert prc_order_bill_new no generated bill_id");
}
String orderNo = syncMapper.selectOrderNo(oldBillId);
UuidToNewIdHolder done = new UuidToNewIdHolder();
done.setOldBillId(oldBillId);
done.setNewBillId(newBillId);
bms-back/bms-system/src/main/java/com/bms/price/task/UuidToNewSyncRunner.java
+61 -6
View File
@@ -1,6 +1,7 @@
package com.bms.price.task;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -10,11 +11,12 @@ import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
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.service.IUuidToNewSyncService;
/**
* UUID 阴影同步运行器：监控页开关控制；
* 开启后立即跑批，一批结束立刻下一批；关闭后当前批结束停。
* 开启后跑批；全失败会退避并自动关闭，避免把库打满。
* 开关/线程/批次落 Redis，服务重启后按 Redis 状态恢复。
*/
@Component
@@ -30,8 +32,14 @@ public class UuidToNewSyncRunner implements ApplicationRunner
/** Redis：批次大小 */
public static final String REDIS_KEY_BATCH_SIZE = "bms:mig:uuid-to-new:batch-size";
/** 连续全失败批次达此值自动关闭 */
private static final int AUTO_OFF_AFTER_ALL_FAIL_BATCHES = 2;
/** 全失败退避基准（毫秒） */
private static final long ALL_FAIL_BACKOFF_MS = 30000L;
private final AtomicBoolean wantRun = new AtomicBoolean(false);
private final AtomicBoolean loopActive = new AtomicBoolean(false);
private final AtomicInteger consecutiveAllFail = new AtomicInteger(0);
@Autowired
private UuidToNewProperties properties;
@@ -62,6 +70,7 @@ public class UuidToNewSyncRunner implements ApplicationRunner
persistEnabled(enabled);
if (enabled)
{
consecutiveAllFail.set(0);
persistRuntimeConfig();
startLoopIfNeeded();
}
@@ -143,6 +152,7 @@ public class UuidToNewSyncRunner implements ApplicationRunner
properties == null ? null : properties.resolvedThreadCount(),
properties == null ? null : properties.resolvedBatchSize());
wantRun.set(true);
consecutiveAllFail.set(0);
startLoopIfNeeded();
}
else
@@ -214,22 +224,57 @@ public class UuidToNewSyncRunner implements ApplicationRunner
while (wantRun.get())
{
int batchSize = properties == null ? 200 : properties.resolvedBatchSize();
int fetched = 0;
UuidToNewBatchResult result = null;
try
{
fetched = syncService.syncBatch(batchSize);
result = syncService.syncBatch(batchSize);
}
catch (Exception e)
{
log.error("uuid-to-new batch error", e);
result = null;
}
if (!wantRun.get())
{
break;
}
if (fetched <= 0)
long sleepMs = properties == null ? 3000L : Math.max(500L, properties.getIdleSleepMs());
if (result == null)
{
sleepMs = Math.max(sleepMs, ALL_FAIL_BACKOFF_MS);
bumpAllFailAndMaybeAutoOff("batch-exception");
}
else if (result.getFetched() <= 0)
{
consecutiveAllFail.set(0);
}
else if (result.getSuccess() <= 0 && result.getFail() > 0)
{
// 全失败：退避，避免立即重打
sleepMs = Math.max(sleepMs, ALL_FAIL_BACKOFF_MS);
bumpAllFailAndMaybeAutoOff(result.getSampleError());
}
else
{
consecutiveAllFail.set(0);
// 有成功：不强制睡眠，但若失败率过高仍稍停
if (result.getFail() > 0 && result.getFail() * 2 >= result.getFetched())
{
sleepMs = Math.max(sleepMs, 5000L);
}
else
{
sleepMs = 0L;
}
}
if (!wantRun.get())
{
break;
}
if (sleepMs > 0L)
{
long sleepMs = properties == null ? 3000L : Math.max(500L, properties.getIdleSleepMs());
try
{
Thread.sleep(sleepMs);
@@ -240,7 +285,6 @@ public class UuidToNewSyncRunner implements ApplicationRunner
break;
}
}
// fetched > 0：立刻下一批
}
}
finally
@@ -249,4 +293,15 @@ public class UuidToNewSyncRunner implements ApplicationRunner
log.info("uuid-to-new loop stopped");
}
}
private void bumpAllFailAndMaybeAutoOff(String sampleError)
{
int n = consecutiveAllFail.incrementAndGet();
log.error("uuid-to-new all-fail batch #{}, sampleErr={}", n, sampleError);
if (n >= AUTO_OFF_AFTER_ALL_FAIL_BATCHES)
{
log.error("uuid-to-new auto switch OFF after {} consecutive all-fail batches", n);
setEnabled(false);
}
}
}
bms-back/bms-system/src/main/resources/mapper/price/UuidToNewSyncMapper.xml
+8
View File
@@ -85,6 +85,14 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
where old_bill_id = #{oldBillId}
</update>
<update id="updateOrderDoneFail">
update mig_bi_order_done
set status = 'FAIL',
fail_reason = #{failReason},
updated_at = now()
where old_bill_id = #{oldBillId}
</update>
<insert id="insertOrderExtNew">
insert into prc_order_bill_ext_new (
bill_id, order_no, create_time, update_time,
bms-front/src/views/monitor/uuidToNew/index.vue
+1 -1
View File
@@ -86,7 +86,7 @@
:stroke-width="14"
:text-inside="true"
/>
<div class="hint-line">关闭开关后可改线程/批次；打开后按当前选项立即连续执行，一批结束立刻下一批。开关/线程/批次会记入 Redis，服务重启后自动恢复。</div>
<div class="hint-line">关闭开关后可改线程/批次；打开后按当前选项连续执行。开关/线程/批次记入 Redis。若连续全失败会退避并自动关闭，避免打满数据库。</div>
</el-card>
<basic-table