From 358e5a46b9fa9274ad4bf4eed3d5e3cd3ac1ff0d Mon Sep 17 00:00:00 2001 From: wuweihua <88888888> Date: Sat, 14 Nov 2026 19:52:38 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9EUUID=E5=88=B0=E6=96=B0?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=E6=9C=8D=E5=8A=A1=E7=9A=84=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E5=A4=84=E7=90=86=E9=80=BB=E8=BE=91=E5=92=8C=E6=89=B9=E6=AC=A1?= =?UTF-8?q?=E7=BB=93=E6=9E=9C=E8=BF=94=E5=9B=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../domain/UuidToNewBatchResult.java | 25 ++++ .../migration/mapper/UuidToNewSyncMapper.java | 2 + .../service/IUuidToNewSyncService.java | 12 +- .../impl/UuidToNewSyncServiceImpl.java | 139 +++++++++++++++--- .../bms/price/task/UuidToNewSyncRunner.java | 67 ++++++++- .../mapper/price/UuidToNewSyncMapper.xml | 8 + .../src/views/monitor/uuidToNew/index.vue | 2 +- 7 files changed, 223 insertions(+), 32 deletions(-) create mode 100644 bms-back/bms-system/src/main/java/com/bms/price/migration/domain/UuidToNewBatchResult.java diff --git a/bms-back/bms-system/src/main/java/com/bms/price/migration/domain/UuidToNewBatchResult.java b/bms-back/bms-system/src/main/java/com/bms/price/migration/domain/UuidToNewBatchResult.java new file mode 100644 index 00000000..fc4b9736 --- /dev/null +++ b/bms-back/bms-system/src/main/java/com/bms/price/migration/domain/UuidToNewBatchResult.java @@ -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; } +} diff --git a/bms-back/bms-system/src/main/java/com/bms/price/migration/mapper/UuidToNewSyncMapper.java b/bms-back/bms-system/src/main/java/com/bms/price/migration/mapper/UuidToNewSyncMapper.java index 7fa8f54f..e3de25a1 100644 --- a/bms-back/bms-system/src/main/java/com/bms/price/migration/mapper/UuidToNewSyncMapper.java +++ b/bms-back/bms-system/src/main/java/com/bms/price/migration/mapper/UuidToNewSyncMapper.java @@ -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 selectIncomeIds(@Param("oldBillId") String oldBillId); diff --git a/bms-back/bms-system/src/main/java/com/bms/price/migration/service/IUuidToNewSyncService.java b/bms-back/bms-system/src/main/java/com/bms/price/migration/service/IUuidToNewSyncService.java index 2c4bf27e..6d07871d 100644 --- a/bms-back/bms-system/src/main/java/com/bms/price/migration/service/IUuidToNewSyncService.java +++ b/bms-back/bms-system/src/main/java/com/bms/price/migration/service/IUuidToNewSyncService.java @@ -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); } diff --git a/bms-back/bms-system/src/main/java/com/bms/price/migration/service/impl/UuidToNewSyncServiceImpl.java b/bms-back/bms-system/src/main/java/com/bms/price/migration/service/impl/UuidToNewSyncServiceImpl.java index 30facab5..67dbc3ed 100644 --- a/bms-back/bms-system/src/main/java/com/bms/price/migration/service/impl/UuidToNewSyncServiceImpl.java +++ b/bms-back/bms-system/src/main/java/com/bms/price/migration/service/impl/UuidToNewSyncServiceImpl.java @@ -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 sampleError = new AtomicReference(); 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 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); diff --git a/bms-back/bms-system/src/main/java/com/bms/price/task/UuidToNewSyncRunner.java b/bms-back/bms-system/src/main/java/com/bms/price/task/UuidToNewSyncRunner.java index 2d17146a..233b1046 100644 --- a/bms-back/bms-system/src/main/java/com/bms/price/task/UuidToNewSyncRunner.java +++ b/bms-back/bms-system/src/main/java/com/bms/price/task/UuidToNewSyncRunner.java @@ -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); + } + } } diff --git a/bms-back/bms-system/src/main/resources/mapper/price/UuidToNewSyncMapper.xml b/bms-back/bms-system/src/main/resources/mapper/price/UuidToNewSyncMapper.xml index 1d35ee29..f9fc9ded 100644 --- a/bms-back/bms-system/src/main/resources/mapper/price/UuidToNewSyncMapper.xml +++ b/bms-back/bms-system/src/main/resources/mapper/price/UuidToNewSyncMapper.xml @@ -85,6 +85,14 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" where old_bill_id = #{oldBillId} + + update mig_bi_order_done + set status = 'FAIL', + fail_reason = #{failReason}, + updated_at = now() + where old_bill_id = #{oldBillId} + + insert into prc_order_bill_ext_new ( bill_id, order_no, create_time, update_time, diff --git a/bms-front/src/views/monitor/uuidToNew/index.vue b/bms-front/src/views/monitor/uuidToNew/index.vue index ba15474f..a4aacdbc 100644 --- a/bms-front/src/views/monitor/uuidToNew/index.vue +++ b/bms-front/src/views/monitor/uuidToNew/index.vue @@ -86,7 +86,7 @@ :stroke-width="14" :text-inside="true" /> -
关闭开关后可改线程/批次；打开后按当前选项立即连续执行，一批结束立刻下一批。开关/线程/批次会记入 Redis，服务重启后自动恢复。
+
关闭开关后可改线程/批次；打开后按当前选项连续执行。开关/线程/批次记入 Redis。若连续全失败会退避并自动关闭，避免打满数据库。