# Please enter a commit message to explain why this merge is necessary,
# especially if it merges an updated upstream into a topic branch.
#
# Lines starting with '#' will be ignored, and an empty message aborts
# the commit.
This commit is contained in:
wuweihua
2026-11-14 19:55:06 +08:00
7 changed files with 223 additions and 32 deletions
@@ -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; }
}
@@ -27,6 +27,8 @@ public interface UuidToNewSyncMapper
int updateOrderDoneSuccess(UuidToNewIdHolder holder); int updateOrderDoneSuccess(UuidToNewIdHolder holder);
int updateOrderDoneFail(UuidToNewIdHolder holder);
int insertOrderExtNew(@Param("oldBillId") String oldBillId, @Param("newBillId") Long newBillId); int insertOrderExtNew(@Param("oldBillId") String oldBillId, @Param("newBillId") Long newBillId);
List<String> selectIncomeIds(@Param("oldBillId") String oldBillId); List<String> selectIncomeIds(@Param("oldBillId") String oldBillId);
@@ -1,5 +1,7 @@
package com.bms.price.migration.service; package com.bms.price.migration.service;
import com.bms.price.migration.domain.UuidToNewBatchResult;
/** /**
* UUID -> bigint 阴影同步服务 * UUID -> bigint 阴影同步服务
*/ */
@@ -7,10 +9,16 @@ public interface IUuidToNewSyncService
{ {
/** /**
* 执行一批。 * 执行一批。
* @return 本批抓取条数(fetched) * @return 本批统计(抓取/成功/失败/跳过)
*/ */
int syncBatch(int batchSize); UuidToNewBatchResult syncBatch(int batchSize);
/** 一单一事务;返回 SUCCESS / FAIL / SKIP */ /** 一单一事务;返回 SUCCESS / FAIL / SKIP */
String syncOneOrder(String oldBillId); String syncOneOrder(String oldBillId);
/** 独立事务占坑;false=已被占用 */
boolean tryClaimOrder(String oldBillId);
/** 独立事务标记 FAIL,避免失败回滚后被无限重试 */
void markOrderFail(String oldBillId, String failReason);
} }
@@ -11,20 +11,23 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future; import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy; import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import com.bms.price.config.UuidToNewProperties; 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.UuidToNewFeeRow;
import com.bms.price.migration.domain.UuidToNewIdHolder; import com.bms.price.migration.domain.UuidToNewIdHolder;
import com.bms.price.migration.mapper.UuidToNewSyncMapper; import com.bms.price.migration.mapper.UuidToNewSyncMapper;
import com.bms.price.migration.service.IUuidToNewSyncService; import com.bms.price.migration.service.IUuidToNewSyncService;
/** /**
* UUID -> bigint 阴影同步:一单一事务;批内可多线程 * UUID -> bigint 阴影同步:占坑独立提交;业务拷贝一单一事务;批内可多线程
*/ */
@Service @Service
public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
@@ -47,7 +50,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
private IUuidToNewSyncService self; private IUuidToNewSyncService self;
@Override @Override
public int syncBatch(int batchSize) public UuidToNewBatchResult syncBatch(int batchSize)
{ {
int limit = Math.max(1, Math.min(batchSize, 1000)); int limit = Math.max(1, Math.min(batchSize, 1000));
int threads = properties == null ? 1 : properties.resolvedThreadCount(); int threads = properties == null ? 1 : properties.resolvedThreadCount();
@@ -60,6 +63,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
AtomicInteger success = new AtomicInteger(0); AtomicInteger success = new AtomicInteger(0);
AtomicInteger fail = new AtomicInteger(0); AtomicInteger fail = new AtomicInteger(0);
AtomicInteger skip = new AtomicInteger(0); AtomicInteger skip = new AtomicInteger(0);
AtomicReference<String> sampleError = new AtomicReference<String>();
if (billIds.isEmpty()) if (billIds.isEmpty())
{ {
@@ -69,7 +73,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
{ {
for (String oldBillId : billIds) for (String oldBillId : billIds)
{ {
applyResult(self, oldBillId, success, fail, skip); applyResult(self, oldBillId, success, fail, skip, sampleError);
} }
} }
else else
@@ -86,7 +90,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
@Override @Override
public void run() 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) catch (Exception e)
{ {
fail.incrementAndGet(); fail.incrementAndGet();
sampleError.compareAndSet(null, abbreviate(e));
log.warn("uuid-to-new worker future error: {}", e.toString()); log.warn("uuid-to-new worker future error: {}", e.toString());
} }
} }
@@ -125,17 +130,29 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
job.setSuccess(success.get()); job.setSuccess(success.get());
job.setFail(fail.get()); job.setFail(fail.get());
job.setSkip(skip.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); syncMapper.updateJobLogFinish(job);
String summary = String.format(
"batch=%s threads=%d fetched=%d success=%d fail=%d skip=%d", UuidToNewBatchResult result = new UuidToNewBatchResult();
batchNo, threads, billIds.size(), success.get(), fail.get(), skip.get()); result.setFetched(billIds.size());
log.info("uuid-to-new {}", summary); result.setSuccess(success.get());
return billIds.size(); 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, private void applyResult(IUuidToNewSyncService svc, String oldBillId,
AtomicInteger success, AtomicInteger fail, AtomicInteger skip) AtomicInteger success, AtomicInteger fail, AtomicInteger skip,
AtomicReference<String> sampleError)
{ {
try try
{ {
@@ -148,6 +165,10 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
{ {
skip.incrementAndGet(); skip.incrementAndGet();
} }
else if (FAIL.equals(r))
{
fail.incrementAndGet();
}
else else
{ {
fail.incrementAndGet(); fail.incrementAndGet();
@@ -156,6 +177,16 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
catch (Exception e) catch (Exception e)
{ {
fail.incrementAndGet(); 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()); log.warn("uuid-to-new sync fail billId={}: {}", oldBillId, e.toString());
} }
} }
@@ -177,6 +208,78 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
return h; 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 @Override
@Transactional(rollbackFor = Exception.class) @Transactional(rollbackFor = Exception.class)
public String syncOneOrder(String oldBillId) public String syncOneOrder(String oldBillId)
@@ -185,18 +288,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
{ {
return SKIP; return SKIP;
} }
if (syncMapper.countOrderBill(oldBillId) <= 0) if (!self.tryClaimOrder(oldBillId))
{
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)
{ {
return SKIP; return SKIP;
} }
@@ -215,6 +307,7 @@ public class UuidToNewSyncServiceImpl implements IUuidToNewSyncService
throw new IllegalStateException("insert prc_order_bill_new no generated bill_id"); throw new IllegalStateException("insert prc_order_bill_new no generated bill_id");
} }
String orderNo = syncMapper.selectOrderNo(oldBillId);
UuidToNewIdHolder done = new UuidToNewIdHolder(); UuidToNewIdHolder done = new UuidToNewIdHolder();
done.setOldBillId(oldBillId); done.setOldBillId(oldBillId);
done.setNewBillId(newBillId); done.setNewBillId(newBillId);
@@ -1,6 +1,7 @@
package com.bms.price.task; package com.bms.price.task;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
@@ -10,11 +11,12 @@ import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import com.bms.common.core.redis.RedisCache; import com.bms.common.core.redis.RedisCache;
import com.bms.price.config.UuidToNewProperties; import com.bms.price.config.UuidToNewProperties;
import com.bms.price.migration.domain.UuidToNewBatchResult;
import com.bms.price.migration.service.IUuidToNewSyncService; import com.bms.price.migration.service.IUuidToNewSyncService;
/** /**
* UUID 阴影同步运行器:监控页开关控制; * UUID 阴影同步运行器:监控页开关控制;
* 开启后立即跑批,一批结束立刻下一批;关闭后当前批结束停。 * 开启后跑批;全失败会退避并自动关闭,避免把库打满。
* 开关/线程/批次落 Redis,服务重启后按 Redis 状态恢复。 * 开关/线程/批次落 Redis,服务重启后按 Redis 状态恢复。
*/ */
@Component @Component
@@ -30,8 +32,14 @@ public class UuidToNewSyncRunner implements ApplicationRunner
/** Redis:批次大小 */ /** Redis:批次大小 */
public static final String REDIS_KEY_BATCH_SIZE = "bms:mig:uuid-to-new:batch-size"; 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 wantRun = new AtomicBoolean(false);
private final AtomicBoolean loopActive = new AtomicBoolean(false); private final AtomicBoolean loopActive = new AtomicBoolean(false);
private final AtomicInteger consecutiveAllFail = new AtomicInteger(0);
@Autowired @Autowired
private UuidToNewProperties properties; private UuidToNewProperties properties;
@@ -62,6 +70,7 @@ public class UuidToNewSyncRunner implements ApplicationRunner
persistEnabled(enabled); persistEnabled(enabled);
if (enabled) if (enabled)
{ {
consecutiveAllFail.set(0);
persistRuntimeConfig(); persistRuntimeConfig();
startLoopIfNeeded(); startLoopIfNeeded();
} }
@@ -143,6 +152,7 @@ public class UuidToNewSyncRunner implements ApplicationRunner
properties == null ? null : properties.resolvedThreadCount(), properties == null ? null : properties.resolvedThreadCount(),
properties == null ? null : properties.resolvedBatchSize()); properties == null ? null : properties.resolvedBatchSize());
wantRun.set(true); wantRun.set(true);
consecutiveAllFail.set(0);
startLoopIfNeeded(); startLoopIfNeeded();
} }
else else
@@ -214,22 +224,57 @@ public class UuidToNewSyncRunner implements ApplicationRunner
while (wantRun.get()) while (wantRun.get())
{ {
int batchSize = properties == null ? 200 : properties.resolvedBatchSize(); int batchSize = properties == null ? 200 : properties.resolvedBatchSize();
int fetched = 0; UuidToNewBatchResult result = null;
try try
{ {
fetched = syncService.syncBatch(batchSize); result = syncService.syncBatch(batchSize);
} }
catch (Exception e) catch (Exception e)
{ {
log.error("uuid-to-new batch error", e); log.error("uuid-to-new batch error", e);
result = null;
} }
if (!wantRun.get()) if (!wantRun.get())
{ {
break; 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 try
{ {
Thread.sleep(sleepMs); Thread.sleep(sleepMs);
@@ -240,7 +285,6 @@ public class UuidToNewSyncRunner implements ApplicationRunner
break; break;
} }
} }
// fetched > 0:立刻下一批
} }
} }
finally finally
@@ -249,4 +293,15 @@ public class UuidToNewSyncRunner implements ApplicationRunner
log.info("uuid-to-new loop stopped"); 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);
}
}
} }
@@ -85,6 +85,14 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
where old_bill_id = #{oldBillId} where old_bill_id = #{oldBillId}
</update> </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 id="insertOrderExtNew">
insert into prc_order_bill_ext_new ( insert into prc_order_bill_ext_new (
bill_id, order_no, create_time, update_time, bill_id, order_no, create_time, update_time,
@@ -86,7 +86,7 @@
:stroke-width="14" :stroke-width="14"
:text-inside="true" :text-inside="true"
/> />
<div class="hint-line">关闭开关后可改线程/批次;打开后按当前选项立即连续执行,一批结束立刻下一批。开关/线程/批次会记入 Redis,服务重启后自动恢复。</div> <div class="hint-line">关闭开关后可改线程/批次;打开后按当前选项连续执行。开关/线程/批次记入 Redis。若连续全失败会退避并自动关闭,避免打满数据库。</div>
</el-card> </el-card>
<basic-table <basic-table