优化订单列表,多单号查询,支持几万单号查询

This commit is contained in:
luojiaquan
2026-10-16 09:22:13 +08:00
parent 0a7f639138
commit 02481b102c
10 changed files with 655 additions and 32 deletions
@@ -37,4 +37,19 @@ public interface OracleUpstreamOrderMapper
List<Map<String, Object>> selectTabOrderIdsByProduct(
@Param("orderTable") String orderTable,
@Param("query") TabOrderMonitorQuery query);
/**
* 大批量单号查询的单批收集:一批不超过 900 个单号,对三列做 OR 精确匹配,
* 同时叠加客户/时间/单号等值等其他过滤,仅返回 ID + PICKUP_DATE。
* 调用方负责分批、去重、排序与内存分页。
*/
List<Map<String, Object>> selectMatchedOrderKeys(
@Param("orderTable") String orderTable,
@Param("query") TabOrderMonitorQuery query,
@Param("batch") List<String> batch);
/** 按已收集到的订单主键批量查明细(一批为一个内存分页切片)。 */
List<Map<String, Object>> selectTabOrderMonitorByIds(
@Param("orderTable") String orderTable,
@Param("ids") List<String> ids);
}
@@ -0,0 +1,30 @@
package com.bms.price.oracle;
import java.sql.Timestamp;
/**
* 大批量单号查询时,分批从 Oracle 收集到的匹配订单主键与排序字段。
* 注意:TAB_ORDER.ID 为 VARCHAR2(64),按字符串处理,不能转数字。
* PICKUP_DATE 用于与列表保持一致的“提货时间倒序、订单ID倒序”内存排序。
*/
public class MatchedOrderKey
{
private final String id;
private final Timestamp pickupDate;
public MatchedOrderKey(String id, Timestamp pickupDate)
{
this.id = id;
this.pickupDate = pickupDate;
}
public String getId()
{
return id;
}
public Timestamp getPickupDate()
{
return pickupDate;
}
}
@@ -0,0 +1,28 @@
package com.bms.price.oracle;
import java.util.List;
import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.bms.common.annotation.DataSource;
import com.bms.common.enums.DataSourceType;
import com.bms.price.domain.vo.TabOrderMonitorQuery;
import com.bms.price.mapper.OracleUpstreamOrderMapper;
/**
* 大批量单号分批收集的工作单元。独立成 Bean 是为了让 @DataSource 切面
* 在并行工作线程内也能正确切换 SLAVE(Oracle)数据源(ThreadLocal 按线程隔离,
* 切面 around 会在每个工作线程内 set/clear)。
*/
@Component
public class OracleOrderKeyCollectWorker
{
@Autowired
private OracleUpstreamOrderMapper oracleUpstreamOrderMapper;
@DataSource(DataSourceType.SLAVE)
public List<Map<String, Object>> collectBatch(String table, TabOrderMonitorQuery query, List<String> batch)
{
return oracleUpstreamOrderMapper.selectMatchedOrderKeys(table, query, batch);
}
}
@@ -4,9 +4,20 @@ import java.math.BigDecimal;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import jakarta.annotation.PreDestroy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -29,16 +40,101 @@ public class TabOrderMonitorService
{
private static final Logger log = LoggerFactory.getLogger(TabOrderMonitorService.class);
/** Oracle 单个 IN 列表上限 1000,分批收集时每批取 900 留安全余量。 */
public static final int IDENTIFIER_BATCH_SIZE = 900;
/**
* 分批收集并行度:Oracle 从库连接池 maxActive=20,6 路并行与同步任务并发合计仍有余量。
* 实测相同 12000 单号:串行约 18s(冷),6 路并行约 1s。
*/
private static final int COLLECT_PARALLELISM = 6;
/** 相同查询条件的收集结果缓存时间:翻页/重复查询直接复用匹配主键,秒回。 */
private static final long COLLECT_CACHE_TTL_MS = 5L * 60 * 1000;
/** 收集结果缓存条目上限(LRU 淘汰),避免不同单号集合无限占内存。 */
private static final int COLLECT_CACHE_MAX_ENTRIES = 20;
/** 单批 Oracle 查询最长等待(从库 socketTimeout 为 120s,留同样余量)。 */
private static final long COLLECT_BATCH_TIMEOUT_SECONDS = 120L;
@Autowired
private OracleUpstreamOrderMapper oracleUpstreamOrderMapper;
@Autowired
private OracleOrderKeyCollectWorker collectWorker;
@Autowired
private BillingProperties billingProperties;
private final ThreadFactory collectThreadFactory = new ThreadFactory()
{
private final AtomicInteger seq = new AtomicInteger(1);
@Override
public Thread newThread(Runnable r)
{
Thread t = new Thread(r, "oracle-orderkey-collect-" + seq.getAndIncrement());
t.setDaemon(true);
return t;
}
};
private final ExecutorService collectPool =
Executors.newFixedThreadPool(COLLECT_PARALLELISM, collectThreadFactory);
/** 查询指纹 -> 已排序匹配主键(短期 TTL + LRU)。 */
private final Map<String, CacheEntry> collectCache =
Collections.synchronizedMap(new LinkedHashMap<String, CacheEntry>(16, 0.75f, true)
{
private static final long serialVersionUID = 1L;
@Override
protected boolean removeEldestEntry(Map.Entry<String, CacheEntry> eldest)
{
return size() > COLLECT_CACHE_MAX_ENTRIES;
}
});
private static final class CacheEntry
{
private final List<MatchedOrderKey> keys;
private final long expireAt;
private CacheEntry(List<MatchedOrderKey> keys, long expireAt)
{
this.keys = keys;
this.expireAt = expireAt;
}
}
@PreDestroy
public void shutdownCollectPool()
{
collectPool.shutdown();
try
{
if (!collectPool.awaitTermination(5, TimeUnit.SECONDS))
{
collectPool.shutdownNow();
}
}
catch (InterruptedException e)
{
collectPool.shutdownNow();
Thread.currentThread().interrupt();
}
}
@DataSource(DataSourceType.SLAVE)
public long count(TabOrderMonitorQuery query)
{
validate(query);
// 大批量单号:分批收集匹配主键,总数即去重后的主键数
if (hasIdentifiers(query))
{
return doCollectMatchedKeys(query, resolveTable()).size();
}
String table = resolveTable();
try
{
@@ -55,10 +151,24 @@ public class TabOrderMonitorService
}
}
/** Oracle 单条 SQL 的 IN 列表上限;产品检测不走分批收集,单号数需在此以内。 */
public static final int PRODUCT_CHECK_IDENTIFIER_LIMIT = 1000;
private static void ensureProductCheckIdentifierLimit(TabOrderMonitorQuery query)
{
if (query != null && query.getOrderIdentifiers() != null
&& query.getOrderIdentifiers().size() > PRODUCT_CHECK_IDENTIFIER_LIMIT)
{
throw new ServiceException("检测产品最多支持 " + PRODUCT_CHECK_IDENTIFIER_LIMIT
+ " 个单号;单号较多时请直接在订单列表查询后勾选/按条件同步计费");
}
}
@DataSource(DataSourceType.SLAVE)
public List<Map<String, Object>> selectDistinctProductCodes(TabOrderMonitorQuery query)
{
validate(query);
ensureProductCheckIdentifierLimit(query);
String table = resolveTable();
try
{
@@ -85,6 +195,7 @@ public class TabOrderMonitorService
public List<Map<String, Object>> selectOrderIdsByProduct(TabOrderMonitorQuery query)
{
validate(query);
ensureProductCheckIdentifierLimit(query);
if (StringUtils.isEmpty(query.getProductCode()))
{
throw new ServiceException("productCode required");
@@ -113,17 +224,91 @@ public class TabOrderMonitorService
}
}
/**
* 列表查询统一入口:一次数据源调用内返回总数 + 当前页,
* 大批量单号场景只做一次分批收集,避免 count 和 page 重复扫描。
*/
@DataSource(DataSourceType.SLAVE)
public TabOrderPageResult searchPage(TabOrderMonitorQuery query, int pageNum, int pageSize)
{
validate(query);
int[] page = normalizePage(pageNum, pageSize);
int pn = page[0];
int ps = page[1];
String table = resolveTable();
if (hasIdentifiers(query))
{
long t0 = System.currentTimeMillis();
List<MatchedOrderKey> keys = doCollectMatchedKeys(query, table);
List<TabOrderMonitorVo> rows = doPageByKeys(table, keys, pn, ps);
log.info("oracle tab_order identifier page identifiers={} matched={} page={}/{} rows={} costMs={}",
Integer.valueOf(query.getOrderIdentifiers().size()), Integer.valueOf(keys.size()),
Integer.valueOf(pn), Integer.valueOf(ps), Integer.valueOf(rows.size()),
Long.valueOf(System.currentTimeMillis() - t0));
return new TabOrderPageResult(keys.size(), rows);
}
long total;
try
{
PageHelper.clearPage();
total = oracleUpstreamOrderMapper.countTabOrderMonitor(table, query);
}
catch (RuntimeException e)
{
throw new ServiceException("Oracle TAB_ORDER count failed: " + e.getMessage());
}
finally
{
PageHelper.clearPage();
}
return new TabOrderPageResult(total, doSelectRownumPage(table, query, pn, ps));
}
@DataSource(DataSourceType.SLAVE)
public List<TabOrderMonitorVo> selectPage(TabOrderMonitorQuery query, int pageNum, int pageSize)
{
validate(query);
int[] page = normalizePage(pageNum, pageSize);
int pn = page[0];
int ps = page[1];
String table = resolveTable();
int pn = pageNum < 1 ? 1 : pageNum;
int ps = pageSize < 1 ? 10 : pageSize;
if (ps > 200)
// 大批量单号:收集匹配主键后按页切片(兼容直接调用本方法的场景)
if (hasIdentifiers(query))
{
ps = 200;
List<MatchedOrderKey> keys = doCollectMatchedKeys(query, table);
return doPageByKeys(table, keys, pn, ps);
}
return doSelectRownumPage(table, query, pn, ps);
}
/**
* 大批量单号场景:分批收集全部匹配订单主键(去重 + 按提货时间/ID倒序)。
* 供同步任务一次性收集后顺序处理,避免每页重复全量扫描。
*/
@DataSource(DataSourceType.SLAVE)
public List<MatchedOrderKey> collectMatchedOrderKeys(TabOrderMonitorQuery query)
{
validate(query);
if (!hasIdentifiers(query))
{
throw new ServiceException("orderIdentifiers required");
}
return doCollectMatchedKeys(query, resolveTable());
}
/** 同步任务按已收集好的主键切片拉取明细,顺序与 keys 切片一致。 */
@DataSource(DataSourceType.SLAVE)
public List<TabOrderMonitorVo> selectPageByKeys(List<MatchedOrderKey> keySlice)
{
if (keySlice == null || keySlice.isEmpty())
{
return Collections.emptyList();
}
return doPageByKeys(resolveTable(), keySlice, 1, keySlice.size());
}
private List<TabOrderMonitorVo> doSelectRownumPage(String table, TabOrderMonitorQuery query, int pn, int ps)
{
long endRow = (long) pn * ps;
long startRow = endRow - ps;
try
@@ -135,16 +320,7 @@ public class TabOrderMonitorService
query.getCustomerCodes() == null ? 0 : query.getCustomerCodes().size(),
query.getBeginTime(), query.getEndTime(), pn, ps,
rows == null ? 0 : rows.size(), System.currentTimeMillis() - t0);
if (rows == null || rows.isEmpty())
{
return Collections.emptyList();
}
List<TabOrderMonitorVo> list = new ArrayList<TabOrderMonitorVo>(rows.size());
for (Map<String, Object> row : rows)
{
list.add(toVo(row));
}
return list;
return toVoList(rows);
}
catch (RuntimeException e)
{
@@ -156,6 +332,246 @@ public class TabOrderMonitorService
}
}
/**
* 核心:单号按 {@link #IDENTIFIER_BATCH_SIZE} 分批,{@link #COLLECT_PARALLELISM} 路并行
* 查 Oracle(每批 UNION ALL 三列各走索引),Java 侧按 ID 去重,
* 再按 PICKUP_DATE DESC, ID DESC 排序。相同查询条件短期缓存,翻页/重复查询秒回。
*/
private List<MatchedOrderKey> doCollectMatchedKeys(TabOrderMonitorQuery query, String table)
{
List<String> identifiers = query.getOrderIdentifiers();
if (identifiers == null || identifiers.isEmpty())
{
return Collections.emptyList();
}
String fingerprint = collectFingerprint(table, query);
long now = System.currentTimeMillis();
CacheEntry cached = collectCache.get(fingerprint);
if (cached != null && now < cached.expireAt)
{
log.info("oracle tab_order identifier collect cache-hit identifiers={} matched={}",
Integer.valueOf(identifiers.size()), Integer.valueOf(cached.keys.size()));
return new ArrayList<MatchedOrderKey>(cached.keys);
}
List<List<String>> batches = new ArrayList<List<String>>();
for (int i = 0; i < identifiers.size(); i += IDENTIFIER_BATCH_SIZE)
{
batches.add(identifiers.subList(i, Math.min(i + IDENTIFIER_BATCH_SIZE, identifiers.size())));
}
// 提交并行任务(工作线程内由 @DataSource 切面切到 Oracle 数据源)
List<Future<List<Map<String, Object>>>> futures =
new ArrayList<Future<List<Map<String, Object>>>>(batches.size());
for (List<String> batch : batches)
{
futures.add(collectPool.submit(() -> collectWorker.collectBatch(table, query, batch)));
}
// ID -> PICKUP_DATE,跨批去重(同一订单可能多列/多批命中;ID 为 VARCHAR2)
Map<String, Timestamp> unique = new LinkedHashMap<String, Timestamp>();
long t0 = System.currentTimeMillis();
try
{
for (Future<List<Map<String, Object>>> future : futures)
{
List<Map<String, Object>> rows;
try
{
rows = future.get(COLLECT_BATCH_TIMEOUT_SECONDS, TimeUnit.SECONDS);
}
catch (ExecutionException e)
{
Throwable cause = e.getCause() == null ? e : e.getCause();
throw new ServiceException("Oracle TAB_ORDER identifier collect failed: " + cause.getMessage());
}
catch (TimeoutException e)
{
future.cancel(true);
throw new ServiceException("Oracle TAB_ORDER identifier collect timeout");
}
catch (InterruptedException e)
{
Thread.currentThread().interrupt();
throw new ServiceException("Oracle TAB_ORDER identifier collect interrupted");
}
if (rows == null)
{
continue;
}
for (Map<String, Object> row : rows)
{
Object idVal = value(row, "ID");
if (idVal == null)
{
continue;
}
String id = String.valueOf(idVal).trim();
if (id.isEmpty())
{
continue;
}
Object dateVal = value(row, "PICKUP_DATE");
Timestamp pickup = dateVal instanceof Timestamp ? (Timestamp) dateVal : null;
// 同一 ID 跨批重复时保留非空提货时间
if (!unique.containsKey(id) || (unique.get(id) == null && pickup != null))
{
unique.put(id, pickup);
}
}
}
}
finally
{
// 异常时取消尚未完成的批次
for (Future<List<Map<String, Object>>> f : futures)
{
if (!f.isDone())
{
f.cancel(true);
}
}
PageHelper.clearPage();
}
List<MatchedOrderKey> keys = new ArrayList<MatchedOrderKey>(unique.size());
for (Map.Entry<String, Timestamp> e : unique.entrySet())
{
keys.add(new MatchedOrderKey(e.getKey(), e.getValue()));
}
// 与列表 SQL 的 ORDER BY PICKUP_DATE DESC, ORDER_ID DESC 保持一致(Oracle DESC 默认 NULLS FIRST)
keys.sort(Comparator
.comparing(MatchedOrderKey::getPickupDate,
Comparator.nullsFirst(Comparator.<Timestamp>reverseOrder()))
.thenComparing(MatchedOrderKey::getId, Comparator.<String>reverseOrder()));
long cost = System.currentTimeMillis() - t0;
log.info("oracle tab_order identifier collect identifiers={} batches={} parallel={} matched={} costMs={}",
Integer.valueOf(identifiers.size()), Integer.valueOf(batches.size()),
Integer.valueOf(COLLECT_PARALLELISM), Integer.valueOf(keys.size()), Long.valueOf(cost));
// 存不可变快照供翻页/重复查询复用;本次返回可变副本
collectCache.put(fingerprint, new CacheEntry(Collections.unmodifiableList(keys),
System.currentTimeMillis() + COLLECT_CACHE_TTL_MS));
return keys;
}
/** 查询指纹:单号/客户集合顺序无关,叠加表名与其余过滤条件;同指纹才可复用收集结果。 */
private static String collectFingerprint(String table, TabOrderMonitorQuery q)
{
StringBuilder sb = new StringBuilder(table == null ? "" : table).append('|');
appendSorted(sb, q.getOrderIdentifiers());
sb.append('|');
appendSorted(sb, q.getCustomerCodes());
sb.append('|').append(safe(q.getBeginTime())).append('|').append(safe(q.getEndTime()))
.append('|').append(safe(q.getOrderId())).append('|').append(safe(q.getOrgOrderId()))
.append('|').append(safe(q.getTransferOrderId()));
return Integer.toHexString(sb.toString().hashCode());
}
private static void appendSorted(StringBuilder sb, List<String> values)
{
if (values == null || values.isEmpty())
{
return;
}
List<String> copy = new ArrayList<String>(values);
Collections.sort(copy);
for (String v : copy)
{
sb.append(v).append(',');
}
}
private static String safe(String s)
{
return s == null ? "" : s;
}
/** 已排序主键的内存分页切片,并按切片顺序回查明细。 */
private List<TabOrderMonitorVo> doPageByKeys(String table, List<MatchedOrderKey> orderedKeys, int pn, int ps)
{
if (orderedKeys == null || orderedKeys.isEmpty())
{
return Collections.emptyList();
}
int from = (pn - 1) * ps;
if (from >= orderedKeys.size())
{
return Collections.emptyList();
}
List<MatchedOrderKey> slice = orderedKeys.subList(from, Math.min(from + ps, orderedKeys.size()));
List<String> ids = new ArrayList<String>(slice.size());
for (MatchedOrderKey key : slice)
{
ids.add(key.getId());
}
try
{
PageHelper.clearPage();
List<Map<String, Object>> rows = oracleUpstreamOrderMapper.selectTabOrderMonitorByIds(table, ids);
Map<String, Map<String, Object>> rowById = new LinkedHashMap<String, Map<String, Object>>();
if (rows != null)
{
for (Map<String, Object> row : rows)
{
Object idVal = value(row, "ID");
if (idVal == null)
{
continue;
}
rowById.put(String.valueOf(idVal).trim(), row);
}
}
// 回查结果无序,按主键切片顺序重排
List<Map<String, Object>> orderedRows = new ArrayList<Map<String, Object>>(slice.size());
for (MatchedOrderKey key : slice)
{
Map<String, Object> row = rowById.get(key.getId());
if (row != null)
{
orderedRows.add(row);
}
}
return toVoList(orderedRows);
}
catch (RuntimeException e)
{
throw new ServiceException("Oracle TAB_ORDER list by ids failed: " + e.getMessage());
}
finally
{
PageHelper.clearPage();
}
}
private static int[] normalizePage(int pageNum, int pageSize)
{
int pn = pageNum < 1 ? 1 : pageNum;
int ps = pageSize < 1 ? 10 : pageSize;
if (ps > 200)
{
ps = 200;
}
return new int[] { pn, ps };
}
private static boolean hasIdentifiers(TabOrderMonitorQuery query)
{
return query != null && query.getOrderIdentifiers() != null && !query.getOrderIdentifiers().isEmpty();
}
private static List<TabOrderMonitorVo> toVoList(List<Map<String, Object>> rows)
{
if (rows == null || rows.isEmpty())
{
return Collections.emptyList();
}
List<TabOrderMonitorVo> list = new ArrayList<TabOrderMonitorVo>(rows.size());
for (Map<String, Object> row : rows)
{
list.add(toVo(row));
}
return list;
}
private void validate(TabOrderMonitorQuery query)
{
if (query == null)
@@ -0,0 +1,27 @@
package com.bms.price.oracle;
import java.util.List;
import com.bms.price.domain.vo.TabOrderMonitorVo;
/** 订单列表一次查询的总数 + 当前页数据(大批量单号场景一次收集、一次分页)。 */
public class TabOrderPageResult
{
private final long total;
private final List<TabOrderMonitorVo> rows;
public TabOrderPageResult(long total, List<TabOrderMonitorVo> rows)
{
this.total = total;
this.rows = rows;
}
public long getTotal()
{
return total;
}
public List<TabOrderMonitorVo> getRows()
{
return rows;
}
}
@@ -73,8 +73,16 @@ public class TabOrderSyncAsyncRunner
ExecutorService pool = newWorkerPool(threads, "order-source-sync");
try
{
long totalLong = tabOrderMonitorService.count(query);
int total = totalLong > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) totalLong;
// 大批量单号:先一次性分批收集全部匹配主键,之后按切片拉明细,避免每页重复全量扫描
List<com.bms.price.oracle.MatchedOrderKey> matchedKeys = null;
if (query != null && query.getOrderIdentifiers() != null && !query.getOrderIdentifiers().isEmpty())
{
matchedKeys = tabOrderMonitorService.collectMatchedOrderKeys(query);
}
final List<com.bms.price.oracle.MatchedOrderKey> keys = matchedKeys;
int total = keys != null
? keys.size()
: toIntExact(tabOrderMonitorService.count(query));
put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_RUNNING, 0, "starting", total, 0, 0, 0, null);
if (total <= 0)
{
@@ -89,7 +97,21 @@ public class TabOrderSyncAsyncRunner
int pageNum = 1;
while (true)
{
List<TabOrderMonitorVo> rows = tabOrderMonitorService.selectPage(query, pageNum, batch);
List<TabOrderMonitorVo> rows;
if (keys != null)
{
int from = (pageNum - 1) * batch;
if (from >= keys.size())
{
break;
}
rows = tabOrderMonitorService.selectPageByKeys(
keys.subList(from, Math.min(from + batch, keys.size())));
}
else
{
rows = tabOrderMonitorService.selectPage(query, pageNum, batch);
}
if (rows == null || rows.isEmpty())
{
break;
@@ -403,6 +425,11 @@ public class TabOrderSyncAsyncRunner
return redisCache.getCacheObject(CACHE_KEY_PREFIX + taskId);
}
private static int toIntExact(long value)
{
return value > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) value;
}
public static int clampBatch(Integer batchSize)
{
int n = batchSize == null ? DEFAULT_BATCH : batchSize.intValue();