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

Merge branch 'test' of https://git.lionquexpress.com/wuweihua/BMS

Browse Source 
# 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-15 20:47:28 +08:00
parent 2f25a99396 1b43bcca4c
commit 0400f55838
8 changed files with 193 additions and 12 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>
bms-back/sql/uuid_to_bi_shadow/20261114_uuid_to_bi_shadow_dev.md
+1 -1
View File
@@ -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 单单步骤
bms-back/sql/uuid_to_bi_shadow/20261115_add_order_create_bill_index.sql
+18
View File
@@ -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;
bms-front/src/views/report/arProfit/index.vue
+1 -4
View File
@@ -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 } }