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

重构产品价和红线价功能

Browse Source
This commit is contained in:
wuweihua
2026-10-18 14:16:27 +08:00
parent db6f5e882a
commit 60d4614695
84 changed files with 6753 additions and 2843 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bms-back/bms-system/src/main/java/com/bms/price/task/BmsProductImportAsyncRunner.java
+244 -2
View File
@@ -1,7 +1,17 @@
package com.bms.price.task;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -9,11 +19,14 @@ import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import com.bms.common.core.redis.RedisCache;
import com.bms.common.utils.StringUtils;
import com.bms.price.domain.BmsProduct;
import com.bms.price.domain.BmsProductCountry;
import com.bms.price.domain.BmsProductImportTask;
import com.bms.price.domain.CatalogPreviewApplyRequest;
import com.bms.price.service.IBmsProductService;
/**
* Async runner for product Excel import (avoids HTTP timeout on large workbooks).
* Async runner for product Excel import / preview (avoids HTTP timeout on large workbooks).
*/
@Component
public class BmsProductImportAsyncRunner
@@ -23,6 +36,8 @@ public class BmsProductImportAsyncRunner
public static final String CACHE_KEY_PREFIX = "price:product:import:task:";
public static final int TTL_MINUTES = 60;
private final ConcurrentHashMap<String, Object> previewResults = new ConcurrentHashMap<String, Object>();
@Autowired
private IBmsProductService productService;
@@ -69,6 +84,215 @@ public class BmsProductImportAsyncRunner
}
}
/** Background multi-country parse-only preview. One task merges up to 5 workbooks. */
@Async("threadPoolTaskExecutor")
public void runPreviewAll(String taskId, List<byte[]> bytesList, List<String> names)
{
try
{
Map<String, Object> merged = parseCatalogFiles(taskId, bytesList, names, false, null);
putSuccess(taskId, "preview ok", merged);
log.info("async product preview done, taskId={}, files={}", taskId, Integer.valueOf(bytesList == null ? 0 : bytesList.size()));
}
catch (Exception e)
{
log.error("async product preview failed, taskId={}", taskId, e);
previewResults.remove(taskId);
put(taskId, BmsProductImportTask.STATUS_FAIL, 100, "failed", errorMessage(e));
}
}
/** Background single-country parse-only preview. One task merges up to 5 workbooks. */
@Async("threadPoolTaskExecutor")
public void runPreviewCountry(String taskId, List<byte[]> bytesList, List<String> names, String countryCode)
{
try
{
Map<String, Object> merged = parseCatalogFiles(taskId, bytesList, names, true, countryCode);
putSuccess(taskId, "preview ok", merged);
log.info("async product country preview done, taskId={}, country={}, files={}", taskId, countryCode,
Integer.valueOf(bytesList == null ? 0 : bytesList.size()));
}
catch (Exception e)
{
log.error("async product country preview failed, taskId={}, country={}", taskId, countryCode, e);
previewResults.remove(taskId);
put(taskId, BmsProductImportTask.STATUS_FAIL, 100, "failed", errorMessage(e));
}
}
/** Parse workbooks together. Merge stays in upload order so a later file wins. */
private Map<String, Object> parseCatalogFiles(String taskId, List<byte[]> bytesList, List<String> names,
boolean country, String countryCode) throws Exception
{
int n = bytesList == null ? 0 : bytesList.size();
if (n == 0)
{
return null;
}
Map<String, Object> merged = null;
for (int i = 0; i < n; i++)
{
put(taskId, BmsProductImportTask.STATUS_RUNNING, 10 + (70 * i / n), "parsing " + (i + 1) + "/" + n, null);
String filename = names != null && i < names.size() ? names.get(i) : null;
byte[] bytes = bytesList.get(i);
bytesList.set(i, null);
Map<String, Object> one = country
? productService.previewImportCountryCatalog(bytes, filename, countryCode)
: productService.previewImportCatalog(bytes, filename);
merged = productService.mergeCatalogPreview(merged, one);
}
return merged;
}
/** Save preview products in parallel. Each product is its own transaction. */
@Async("threadPoolTaskExecutor")
public void runApplyPreview(String taskId, CatalogPreviewApplyRequest request)
{
ExecutorService pool = null;
try
{
Map<String, BmsProduct> meta = new HashMap<String, BmsProduct>();
if (request.getProducts() != null)
{
for (BmsProduct p : request.getProducts())
{
if (p != null && StringUtils.isNotEmpty(p.getProductCode()))
{
meta.put(p.getProductCode().trim().toUpperCase(Locale.ROOT), p);
}
}
}
List<Map.Entry<String, List<BmsProductCountry>>> jobs =
new ArrayList<Map.Entry<String, List<BmsProductCountry>>>();
Map<String, List<BmsProductCountry>> byCode = request.getDetailsByProductCode();
if (byCode != null)
{
for (Map.Entry<String, List<BmsProductCountry>> e : byCode.entrySet())
{
if (e != null && StringUtils.isNotEmpty(e.getKey()) && e.getValue() != null && !e.getValue().isEmpty())
{
jobs.add(e);
}
}
}
int n = jobs.size();
if (n == 0)
{
put(taskId, BmsProductImportTask.STATUS_SUCCESS, 100, "\u5df2\u4fdd\u5b58", null);
return;
}
put(taskId, BmsProductImportTask.STATUS_RUNNING, 5, "\u6b63\u5728\u4fdd\u5b58 0/" + n, null);
Long psvId = request.getPsvId();
AtomicInteger done = new AtomicInteger();
pool = Executors.newFixedThreadPool(Math.min(4, n));
List<Future<?>> futures = new ArrayList<Future<?>>();
for (Map.Entry<String, List<BmsProductCountry>> e : jobs)
{
final String code = e.getKey().trim().toUpperCase(Locale.ROOT);
final List<BmsProductCountry> details = e.getValue();
final BmsProduct product = meta.get(code);
futures.add(pool.submit(new Runnable()
{
@Override
public void run()
{
applyWithDeadlockRetry(new Runnable()
{
@Override
public void run()
{
productService.applyOneCatalogProduct(psvId, code, product, details);
}
});
int d = done.incrementAndGet();
put(taskId, BmsProductImportTask.STATUS_RUNNING, 5 + (90 * d / n),
"\u6b63\u5728\u4fdd\u5b58 " + d + "/" + n, null);
}
}));
}
for (Future<?> future : futures)
{
future.get();
}
put(taskId, BmsProductImportTask.STATUS_SUCCESS, 100, "\u5df2\u4fdd\u5b58\u5230\u5f85\u751f\u6548\u8349\u7a3f", null);
log.info("async product apply preview done, taskId={}, products={}", taskId, Integer.valueOf(n));
}
catch (Exception e)
{
log.error("async product apply preview failed, taskId={}", taskId, e);
Throwable cause = e.getCause() == null ? e : e.getCause();
String msg = cause.getMessage();
if (StringUtils.isEmpty(msg))
{
msg = cause.getClass().getSimpleName();
}
put(taskId, BmsProductImportTask.STATUS_FAIL, 100, "failed", msg);
}
finally
{
if (pool != null)
{
pool.shutdownNow();
}
}
}
/** InnoDB rolls the product transaction back on deadlock; start a new one. */
private void applyWithDeadlockRetry(Runnable action)
{
int max = 5;
for (int attempt = 1; ; attempt++)
{
try
{
action.run();
return;
}
catch (RuntimeException ex)
{
if (attempt >= max || !isDeadlock(ex))
{
throw ex;
}
log.warn("product catalog save deadlock, retry {}/{}", Integer.valueOf(attempt), Integer.valueOf(max));
try
{
Thread.sleep(50L * attempt);
}
catch (InterruptedException ie)
{
Thread.currentThread().interrupt();
throw ex;
}
}
}
}
private static boolean isDeadlock(Throwable ex)
{
Throwable cur = ex;
while (cur != null)
{
if (cur instanceof org.springframework.dao.PessimisticLockingFailureException)
{
return true;
}
String name = cur.getClass().getName();
if (name.indexOf("MySQLTransactionRollbackException") >= 0)
{
return true;
}
String msg = cur.getMessage();
if (msg != null && msg.indexOf("Deadlock found when trying to get lock") >= 0)
{
return true;
}
cur = cur.getCause();
}
return false;
}
public void put(String taskId, String status, int percent, String message, String error)
{
BmsProductImportTask task = new BmsProductImportTask();
@@ -80,9 +304,27 @@ public class BmsProductImportAsyncRunner
redisCache.setCacheObject(CACHE_KEY_PREFIX + taskId, task, TTL_MINUTES, TimeUnit.MINUTES);
}
private void putSuccess(String taskId, String message, Object result)
{
if (result != null)
{
previewResults.put(taskId, result);
}
put(taskId, BmsProductImportTask.STATUS_SUCCESS, 100, message, null);
}
public BmsProductImportTask get(String taskId)
{
return redisCache.getCacheObject(CACHE_KEY_PREFIX + taskId);
BmsProductImportTask task = redisCache.getCacheObject(CACHE_KEY_PREFIX + taskId);
if (task != null)
{
Object result = previewResults.get(taskId);
if (result != null)
{
task.setResult(result);
}
}
return task;
}
private static String errorMessage(Exception e)
bms-back/bms-system/src/main/java/com/bms/price/task/BmsRedlineProductImportAsyncRunner.java
+248 -6
View File
@@ -1,7 +1,17 @@
package com.bms.price.task;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -9,11 +19,14 @@ import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import com.bms.common.core.redis.RedisCache;
import com.bms.common.utils.StringUtils;
import com.bms.price.domain.BmsRedlineProduct;
import com.bms.price.domain.BmsRedlineProductCountry;
import com.bms.price.domain.BmsRedlineProductImportTask;
import com.bms.price.domain.RedlineCatalogPreviewApplyRequest;
import com.bms.price.service.IBmsRedlineProductService;
/**
* Async runner for product Excel import (avoids HTTP timeout on large workbooks).
* Async runner for redline Excel import / preview (avoids HTTP timeout on large workbooks).
*/
@Component
public class BmsRedlineProductImportAsyncRunner
@@ -23,6 +36,8 @@ public class BmsRedlineProductImportAsyncRunner
public static final String CACHE_KEY_PREFIX = "price:redline:import:task:";
public static final int TTL_MINUTES = 60;
private final ConcurrentHashMap<String, Object> previewResults = new ConcurrentHashMap<String, Object>();
@Autowired
private IBmsRedlineProductService productService;
@@ -40,11 +55,11 @@ public class BmsRedlineProductImportAsyncRunner
String message = productService.importProductExcel(bytes, filename, updateSupport, operName, taskId,
effectiveDate);
put(taskId, BmsRedlineProductImportTask.STATUS_SUCCESS, 100, message, null);
log.info("async product import done, taskId={}", taskId);
log.info("async redline import done, taskId={}", taskId);
}
catch (Exception e)
{
log.error("async product import failed, taskId={}", taskId, e);
log.error("async redline import failed, taskId={}", taskId, e);
put(taskId, BmsRedlineProductImportTask.STATUS_FAIL, 100, "failed", errorMessage(e));
}
}
@@ -60,15 +75,224 @@ public class BmsRedlineProductImportAsyncRunner
String message = productService.importCountryProductExcel(bytes, filename, countryCode, operName, taskId,
effectiveDate);
put(taskId, BmsRedlineProductImportTask.STATUS_SUCCESS, 100, message, null);
log.info("async product country import done, taskId={}, country={}", taskId, countryCode);
log.info("async redline country import done, taskId={}, country={}", taskId, countryCode);
}
catch (Exception e)
{
log.error("async product country import failed, taskId={}, country={}", taskId, countryCode, e);
log.error("async redline country import failed, taskId={}, country={}", taskId, countryCode, e);
put(taskId, BmsRedlineProductImportTask.STATUS_FAIL, 100, "failed", errorMessage(e));
}
}
/** Background multi-country parse-only preview. One task merges up to 5 workbooks. */
@Async("threadPoolTaskExecutor")
public void runPreviewAll(String taskId, List<byte[]> bytesList, List<String> names)
{
try
{
Map<String, Object> merged = parseCatalogFiles(taskId, bytesList, names, false, null);
putSuccess(taskId, "preview ok", merged);
log.info("async redline preview done, taskId={}, files={}", taskId, Integer.valueOf(bytesList == null ? 0 : bytesList.size()));
}
catch (Exception e)
{
log.error("async redline preview failed, taskId={}", taskId, e);
previewResults.remove(taskId);
put(taskId, BmsRedlineProductImportTask.STATUS_FAIL, 100, "failed", errorMessage(e));
}
}
/** Background single-country parse-only preview. One task merges up to 5 workbooks. */
@Async("threadPoolTaskExecutor")
public void runPreviewCountry(String taskId, List<byte[]> bytesList, List<String> names, String countryCode)
{
try
{
Map<String, Object> merged = parseCatalogFiles(taskId, bytesList, names, true, countryCode);
putSuccess(taskId, "preview ok", merged);
log.info("async redline country preview done, taskId={}, country={}, files={}", taskId, countryCode,
Integer.valueOf(bytesList == null ? 0 : bytesList.size()));
}
catch (Exception e)
{
log.error("async redline country preview failed, taskId={}, country={}", taskId, countryCode, e);
previewResults.remove(taskId);
put(taskId, BmsRedlineProductImportTask.STATUS_FAIL, 100, "failed", errorMessage(e));
}
}
/** Parse workbooks together. Merge stays in upload order so a later file wins. */
private Map<String, Object> parseCatalogFiles(String taskId, List<byte[]> bytesList, List<String> names,
boolean country, String countryCode) throws Exception
{
int n = bytesList == null ? 0 : bytesList.size();
if (n == 0)
{
return null;
}
Map<String, Object> merged = null;
for (int i = 0; i < n; i++)
{
put(taskId, BmsRedlineProductImportTask.STATUS_RUNNING, 10 + (70 * i / n), "parsing " + (i + 1) + "/" + n, null);
String filename = names != null && i < names.size() ? names.get(i) : null;
byte[] bytes = bytesList.get(i);
bytesList.set(i, null);
Map<String, Object> one = country
? productService.previewImportCountryCatalog(bytes, filename, countryCode)
: productService.previewImportCatalog(bytes, filename);
merged = productService.mergeCatalogPreview(merged, one);
}
return merged;
}
/** Save preview products in parallel. Each product is its own transaction. */
@Async("threadPoolTaskExecutor")
public void runApplyPreview(String taskId, RedlineCatalogPreviewApplyRequest request)
{
ExecutorService pool = null;
try
{
Map<String, BmsRedlineProduct> meta = new HashMap<String, BmsRedlineProduct>();
if (request.getProducts() != null)
{
for (BmsRedlineProduct p : request.getProducts())
{
if (p != null && StringUtils.isNotEmpty(p.getProductCode()))
{
meta.put(p.getProductCode().trim().toUpperCase(Locale.ROOT), p);
}
}
}
List<Map.Entry<String, List<BmsRedlineProductCountry>>> jobs =
new ArrayList<Map.Entry<String, List<BmsRedlineProductCountry>>>();
Map<String, List<BmsRedlineProductCountry>> byCode = request.getDetailsByProductCode();
if (byCode != null)
{
for (Map.Entry<String, List<BmsRedlineProductCountry>> e : byCode.entrySet())
{
if (e != null && StringUtils.isNotEmpty(e.getKey()) && e.getValue() != null && !e.getValue().isEmpty())
{
jobs.add(e);
}
}
}
int n = jobs.size();
if (n == 0)
{
put(taskId, BmsRedlineProductImportTask.STATUS_SUCCESS, 100, "\u5df2\u4fdd\u5b58", null);
return;
}
put(taskId, BmsRedlineProductImportTask.STATUS_RUNNING, 5, "\u6b63\u5728\u4fdd\u5b58 0/" + n, null);
Long psvId = request.getPsvId();
AtomicInteger done = new AtomicInteger();
pool = Executors.newFixedThreadPool(Math.min(4, n));
List<Future<?>> futures = new ArrayList<Future<?>>();
for (Map.Entry<String, List<BmsRedlineProductCountry>> e : jobs)
{
final String code = e.getKey().trim().toUpperCase(Locale.ROOT);
final List<BmsRedlineProductCountry> details = e.getValue();
final BmsRedlineProduct product = meta.get(code);
futures.add(pool.submit(new Runnable()
{
@Override
public void run()
{
applyWithDeadlockRetry(new Runnable()
{
@Override
public void run()
{
productService.applyOneCatalogProduct(psvId, code, product, details);
}
});
int d = done.incrementAndGet();
put(taskId, BmsRedlineProductImportTask.STATUS_RUNNING, 5 + (90 * d / n),
"\u6b63\u5728\u4fdd\u5b58 " + d + "/" + n, null);
}
}));
}
for (Future<?> future : futures)
{
future.get();
}
put(taskId, BmsRedlineProductImportTask.STATUS_SUCCESS, 100, "\u5df2\u4fdd\u5b58\u5230\u5f85\u751f\u6548\u8349\u7a3f", null);
log.info("async redline apply preview done, taskId={}, products={}", taskId, Integer.valueOf(n));
}
catch (Exception e)
{
log.error("async redline apply preview failed, taskId={}", taskId, e);
Throwable cause = e.getCause() == null ? e : e.getCause();
String msg = cause.getMessage();
if (StringUtils.isEmpty(msg))
{
msg = cause.getClass().getSimpleName();
}
put(taskId, BmsRedlineProductImportTask.STATUS_FAIL, 100, "failed", msg);
}
finally
{
if (pool != null)
{
pool.shutdownNow();
}
}
}
/** InnoDB rolls the product transaction back on deadlock; start a new one. */
private void applyWithDeadlockRetry(Runnable action)
{
int max = 5;
for (int attempt = 1; ; attempt++)
{
try
{
action.run();
return;
}
catch (RuntimeException ex)
{
if (attempt >= max || !isDeadlock(ex))
{
throw ex;
}
log.warn("redline catalog save deadlock, retry {}/{}", Integer.valueOf(attempt), Integer.valueOf(max));
try
{
Thread.sleep(50L * attempt);
}
catch (InterruptedException ie)
{
Thread.currentThread().interrupt();
throw ex;
}
}
}
}
private static boolean isDeadlock(Throwable ex)
{
Throwable cur = ex;
while (cur != null)
{
if (cur instanceof org.springframework.dao.PessimisticLockingFailureException)
{
return true;
}
String name = cur.getClass().getName();
if (name.indexOf("MySQLTransactionRollbackException") >= 0)
{
return true;
}
String msg = cur.getMessage();
if (msg != null && msg.indexOf("Deadlock found when trying to get lock") >= 0)
{
return true;
}
cur = cur.getCause();
}
return false;
}
public void put(String taskId, String status, int percent, String message, String error)
{
BmsRedlineProductImportTask task = new BmsRedlineProductImportTask();
@@ -80,9 +304,27 @@ public class BmsRedlineProductImportAsyncRunner
redisCache.setCacheObject(CACHE_KEY_PREFIX + taskId, task, TTL_MINUTES, TimeUnit.MINUTES);
}
private void putSuccess(String taskId, String message, Object result)
{
if (result != null)
{
previewResults.put(taskId, result);
}
put(taskId, BmsRedlineProductImportTask.STATUS_SUCCESS, 100, message, null);
}
public BmsRedlineProductImportTask get(String taskId)
{
return redisCache.getCacheObject(CACHE_KEY_PREFIX + taskId);
BmsRedlineProductImportTask task = redisCache.getCacheObject(CACHE_KEY_PREFIX + taskId);
if (task != null)
{
Object result = previewResults.get(taskId);
if (result != null)
{
task.setResult(result);
}
}
return task;
}
private static String errorMessage(Exception e)
bms-back/bms-system/src/main/java/com/bms/price/task/PrcPriceSetSaveAsyncRunner.java
+32
View File
@@ -116,6 +116,38 @@ public class PrcPriceSetSaveAsyncRunner
}
}
/** Import up to 5 workbooks in one task. Each file commits before the next. */
@Async("threadPoolTaskExecutor")
public void runImportPricesFiles(String taskId, Long priceSetId, Long psvId, List<byte[]> bytesList, List<String> names,
String operName, boolean updateOnly, boolean replaceByProductCountry, String countryCode)
{
try
{
int n = bytesList == null ? 0 : bytesList.size();
String message = null;
for (int i = 0; i < n; i++)
{
int pct = 10 + (80 * i / Math.max(n, 1));
put(taskId, PrcPriceSetSaveTask.STATUS_RUNNING, pct, "parsing " + (i + 1) + "/" + n, priceSetId, null);
String filename = names != null && i < names.size() ? names.get(i) : null;
message = priceSetService.importDetailPrices(priceSetId, psvId, bytesList.get(i), filename, operName, taskId,
updateOnly, replaceByProductCountry, countryCode);
}
put(taskId, PrcPriceSetSaveTask.STATUS_SUCCESS, 100, message, priceSetId, null);
log.info("async import prices files done, taskId={}, priceSetId={}, files={}", taskId, priceSetId, Integer.valueOf(n));
}
catch (Exception e)
{
log.error("async import prices files failed, taskId={}, priceSetId={}", taskId, priceSetId, e);
String msg = e.getMessage();
if (StringUtils.isEmpty(msg))
{
msg = e.getClass().getSimpleName();
}
put(taskId, PrcPriceSetSaveTask.STATUS_FAIL, 100, "failed", priceSetId, msg);
}
}
/** Background pull of product library into package version details. */
@Async("threadPoolTaskExecutor")
public void runPullLibrary(String taskId, Long priceSetId, Long psvId, boolean force, String operName)