diff --git a/bms-back/bms-system/src/main/java/com/bms/price/migration/domain/UuidToNewPendingRow.java b/bms-back/bms-system/src/main/java/com/bms/price/migration/domain/UuidToNewPendingRow.java new file mode 100644 index 00000000..908f68d1 --- /dev/null +++ b/bms-back/bms-system/src/main/java/com/bms/price/migration/domain/UuidToNewPendingRow.java @@ -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; } +} 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 b4685735..8725b7d2 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 @@ -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 selectPendingBillIds(@Param("limit") int limit); + /** + * 游标抓取未处理订单（避免每批从最早订单全量排序） + * @param lastCreateTime 上批最后 create_time；空=从头 + * @param lastBillId 上批最后 bill_id；与 time 配对 + */ + List selectPendingBillIds(@Param("limit") int limit, + @Param("lastCreateTime") Date lastCreateTime, + @Param("lastBillId") String lastBillId); + + /** done 已有进度且 Redis 无游标时：用最新 done 边界初始化 */ + UuidToNewPendingRow selectPendingCursorBootstrap(); String selectOrderNo(@Param("oldBillId") String oldBillId); 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 bdb91ceb..b4d5bd48 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 @@ -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 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 pending = syncMapper.selectPendingBillIds(limit, lastCreateTime, lastBillId); + List billIds = new ArrayList(); + 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()); + } + } } 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 233b1046..a5ce436e 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 @@ -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) 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 e3511635..4318f71d 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 @@ -4,15 +4,32 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd"> - + 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 + + and ( + o.create_time > #{lastCreateTime} + or (o.create_time = #{lastCreateTime} and o.bill_id > #{lastBillId}) + ) + + 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} + + diff --git a/bms-back/sql/uuid_to_bi_shadow/20261114_uuid_to_bi_shadow_dev.md b/bms-back/sql/uuid_to_bi_shadow/20261114_uuid_to_bi_shadow_dev.md index ad2ec12d..fab15ca5 100644 --- a/bms-back/sql/uuid_to_bi_shadow/20261114_uuid_to_bi_shadow_dev.md +++ b/bms-back/sql/uuid_to_bi_shadow/20261114_uuid_to_bi_shadow_dev.md @@ -91,7 +91,7 @@ | 批次 | `batch-size`（建议 200） | | 事务 | **一单一事务**（订单+income 插入 + 关联表 UPDATE order_id） | | 失败 | 整单回滚；FAIL 落 `mig_bi_order_done`（独立事务），避免死循环 | -| Redis | 开关/线程/批次可持久；建议启动不自动开（可再讨） | +| Redis | 开关/线程/批次/抓取游标可持久；游标 key：cursor-time / cursor-bill-id | ### 4.2 单单步骤 diff --git a/bms-back/sql/uuid_to_bi_shadow/20261115_add_order_create_bill_index.sql b/bms-back/sql/uuid_to_bi_shadow/20261115_add_order_create_bill_index.sql new file mode 100644 index 00000000..94fa6492 --- /dev/null +++ b/bms-back/sql/uuid_to_bi_shadow/20261115_add_order_create_bill_index.sql @@ -0,0 +1,18 @@ +-- 游标抓取加速：prc_order_bill (create_time, bill_id) +-- 可重入 + +SET NAMES utf8mb4; + +SET @idx_exists := ( + SELECT COUNT(*) FROM information_schema.statistics + WHERE table_schema = DATABASE() + AND table_name = 'prc_order_bill' + AND index_name = 'idx_order_bill_create_bill' +); + +SET @sql := IF(@idx_exists = 0, + 'ALTER TABLE `prc_order_bill` ADD INDEX `idx_order_bill_create_bill` (`create_time`, `bill_id`)', + 'SELECT ''idx_order_bill_create_bill already exists'''); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; diff --git a/bms-front/src/views/report/arProfit/index.vue b/bms-front/src/views/report/arProfit/index.vue index aaf2ade4..2eb8b23e 100644 --- a/bms-front/src/views/report/arProfit/index.vue +++ b/bms-front/src/views/report/arProfit/index.vue @@ -244,7 +244,6 @@ function loadRows() { const version = ++requestVersion loading.value = true rows.value = [] - total.value = 0 listArProfitBills({ ...params(), pageNum: pageNum.value, pageSize: pageSize.value }).then(res => { if (version === requestVersion) { rows.value = res.rows || []; total.value = res.total || 0 } }).finally(() => { if (version === requestVersion) loading.value = false }) @@ -260,7 +259,6 @@ async function loadBillDetails() { const version = ++detailsVersion detailsLoading.value = true details.value = [] - detailsTotal.value = 0 try { const res = await listArProfitBillDetails(detailBillId.value, { pageNum: detailsPage.value, pageSize: detailsSize.value }); if (version === detailsVersion) { details.value = res.rows || []; detailsTotal.value = res.total || 0 } } finally { if (version === detailsVersion) detailsLoading.value = false } } @@ -285,7 +283,6 @@ async function loadBillCommission() { billCommissionLoading.value = true billCommissionResults.value = [] billCommissionSummary.value = [] - billCommissionTotal.value = 0 try { const [result, summary] = await Promise.all([ listArProfitCommissions({ billId: currentBillId, pageNum: billCommissionPage.value, pageSize: billCommissionSize.value }), @@ -304,7 +301,7 @@ async function loadCommissionTab() { const version = ++commissionVersion const tab = commissionTab.value commissionLoading.value = true - if (tab === 'results') { results.value = []; commissionSummary.value = []; commissionTotal.value = 0 } + if (tab === 'results') { results.value = []; commissionSummary.value = [] } try { if (tab === 'rates') { const r = await listArProfitRates(); if (version === commissionVersion) rates.value = r.data || [] } else if (tab === 'missing') { const r = await listArProfitMissingRates({ pageNum: missingPage.value, pageSize: missingSize.value }); if (version === commissionVersion) { missingRates.value = r.rows || []; missingTotal.value = r.total || 0 } }