增加上传附件到oss
This commit is contained in:
+26
-7
@@ -18,6 +18,7 @@ import com.bms.common.core.domain.AjaxResult;
|
||||
import com.bms.common.utils.StringUtils;
|
||||
import com.bms.common.utils.file.FileUploadUtils;
|
||||
import com.bms.common.utils.file.FileUtils;
|
||||
import com.bms.common.utils.file.OssStorage;
|
||||
import com.bms.framework.config.ServerConfig;
|
||||
|
||||
/**
|
||||
@@ -34,6 +35,9 @@ public class CommonController
|
||||
@Autowired
|
||||
private ServerConfig serverConfig;
|
||||
|
||||
@Autowired
|
||||
private OssStorage ossStorage;
|
||||
|
||||
private static final String FILE_DELIMITER = ",";
|
||||
|
||||
/**
|
||||
@@ -132,7 +136,7 @@ public class CommonController
|
||||
}
|
||||
|
||||
/**
|
||||
* 本地资源通用下载
|
||||
* Common resource download: local profile path or oss://bucket/key (AR bill style).
|
||||
*/
|
||||
@GetMapping("/download/resource")
|
||||
public void resourceDownload(String resource, HttpServletRequest request, HttpServletResponse response)
|
||||
@@ -140,15 +144,30 @@ public class CommonController
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!FileUtils.checkAllowDownload(resource))
|
||||
if (StringUtils.isEmpty(resource))
|
||||
{
|
||||
throw new Exception(StringUtils.format("资源文件({})非法,不允许下载。 ", resource));
|
||||
throw new Exception("resource empty");
|
||||
}
|
||||
boolean ossFile = resource.startsWith("oss://");
|
||||
if (!ossFile && !FileUtils.checkAllowDownload(resource))
|
||||
{
|
||||
throw new Exception(StringUtils.format("resource ({}) not allowed", resource));
|
||||
}
|
||||
if (ossFile)
|
||||
{
|
||||
// Extension check still applies to the object key suffix.
|
||||
if (!FileUtils.checkAllowDownload(resource))
|
||||
{
|
||||
throw new Exception(StringUtils.format("resource ({}) not allowed", resource));
|
||||
}
|
||||
String downloadName = StringUtils.substringAfterLast(resource, "/");
|
||||
response.setContentType(MediaType.APPLICATION_OCTET_STREAM_VALUE);
|
||||
FileUtils.setAttachmentResponseHeader(response, downloadName);
|
||||
ossStorage.download(resource, response.getOutputStream());
|
||||
return;
|
||||
}
|
||||
// 本地资源路径
|
||||
String localPath = BmsConfig.getProfile();
|
||||
// 数据库资源地址
|
||||
String downloadPath = localPath + FileUtils.stripPrefix(resource);
|
||||
// 下载名称
|
||||
String downloadName = StringUtils.substringAfterLast(downloadPath, "/");
|
||||
response.setContentType(MediaType.APPLICATION_OCTET_STREAM_VALUE);
|
||||
FileUtils.setAttachmentResponseHeader(response, downloadName);
|
||||
@@ -156,7 +175,7 @@ public class CommonController
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
log.error("下载文件失败", e);
|
||||
log.error("download resource failed", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+109
-13
@@ -22,6 +22,8 @@ import org.apache.poi.ss.usermodel.Row;
|
||||
import org.apache.poi.ss.usermodel.Sheet;
|
||||
import org.apache.poi.ss.usermodel.Workbook;
|
||||
import org.apache.poi.ss.usermodel.WorkbookFactory;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
@@ -66,6 +68,8 @@ import com.bms.system.mapper.BmsRegionMapper;
|
||||
@Service
|
||||
public class BmsProductServiceImpl implements IBmsProductService
|
||||
{
|
||||
private static final Logger log = LoggerFactory.getLogger(BmsProductServiceImpl.class);
|
||||
|
||||
private static final String CATALOG_SHEET = "\u4ea7\u54c1\u76ee\u5f55";
|
||||
private static final Set<String> SKIP_SHEETS = new HashSet<>(Arrays.asList(
|
||||
CATALOG_SHEET, "\u9000\u4ef6\u91cd\u6d3e", "\u7533\u62a5\u8bf4\u660e", "\u8d54\u507f\u6807\u51c6"));
|
||||
@@ -415,26 +419,16 @@ public class BmsProductServiceImpl implements IBmsProductService
|
||||
{
|
||||
index++;
|
||||
reportImportProductProgress(taskId, index, total, product.getProductCode());
|
||||
DefaultTransactionDefinition def = new DefaultTransactionDefinition();
|
||||
def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
|
||||
TransactionStatus tx = transactionManager.getTransaction(def);
|
||||
try
|
||||
{
|
||||
product.setEffectiveDate(forcedEff);
|
||||
if (importPsvId != null)
|
||||
{
|
||||
product.setPsvId(importPsvId);
|
||||
}
|
||||
int[] nums = importOneProduct(workbook, product, operName, regionByNameCn, euRegions,
|
||||
zoneByCountryAndCode, unmatchedCountries, forcedEff);
|
||||
transactionManager.commit(tx);
|
||||
int[] nums = importOneProductWithDeadlockRetry(workbook, product, operName, regionByNameCn,
|
||||
euRegions, zoneByCountryAndCode, unmatchedCountries, forcedEff, importPsvId);
|
||||
successNum += nums[0];
|
||||
skipNum += nums[1];
|
||||
countryNum += nums[2];
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
transactionManager.rollback(tx);
|
||||
String tip = e.getMessage();
|
||||
if (StringUtils.isEmpty(tip))
|
||||
{
|
||||
@@ -1151,6 +1145,7 @@ public class BmsProductServiceImpl implements IBmsProductService
|
||||
d.setIsDel(BmsProduct.IS_DEL_NO);
|
||||
toInsert.add(d);
|
||||
}
|
||||
sortProductCountriesForInsert(toInsert);
|
||||
insertProductCountryChunks(toInsert);
|
||||
return toInsert.size();
|
||||
}
|
||||
@@ -1169,13 +1164,36 @@ public class BmsProductServiceImpl implements IBmsProductService
|
||||
return taskId;
|
||||
}
|
||||
|
||||
/** Stable insert order reduces InnoDB secondary-index lock inversion. */
|
||||
private static void sortProductCountriesForInsert(List<BmsProductCountry> rows)
|
||||
{
|
||||
if (rows == null || rows.size() < 2)
|
||||
{
|
||||
return;
|
||||
}
|
||||
rows.sort((a, b) -> {
|
||||
int c = String.valueOf(a == null ? null : a.getCountryCode())
|
||||
.compareToIgnoreCase(String.valueOf(b == null ? null : b.getCountryCode()));
|
||||
if (c != 0) return c;
|
||||
c = String.valueOf(a == null ? null : a.getZoneCode())
|
||||
.compareToIgnoreCase(String.valueOf(b == null ? null : b.getZoneCode()));
|
||||
if (c != 0) return c;
|
||||
java.math.BigDecimal aw = a == null ? null : a.getWeightMin();
|
||||
java.math.BigDecimal bw = b == null ? null : b.getWeightMin();
|
||||
if (aw == null && bw == null) return 0;
|
||||
if (aw == null) return -1;
|
||||
if (bw == null) return 1;
|
||||
return aw.compareTo(bw);
|
||||
});
|
||||
}
|
||||
|
||||
private void insertProductCountryChunks(List<BmsProductCountry> rows)
|
||||
{
|
||||
if (rows == null || rows.isEmpty())
|
||||
{
|
||||
return;
|
||||
}
|
||||
int size = 200;
|
||||
int size = 100;
|
||||
for (int i = 0; i < rows.size(); i += size)
|
||||
{
|
||||
int end = Math.min(i + size, rows.size());
|
||||
@@ -1183,6 +1201,84 @@ public class BmsProductServiceImpl implements IBmsProductService
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* One-product import in REQUIRES_NEW; retry whole txn on InnoDB deadlock.
|
||||
*/
|
||||
private int[] importOneProductWithDeadlockRetry(Workbook workbook, BmsProduct product, String operName,
|
||||
Map<String, BmsRegion> regionByNameCn, List<BmsRegion> euRegions,
|
||||
Map<String, PrcZone> zoneByCountryAndCode, Set<String> unmatchedCountries, Date forcedEff, Long importPsvId)
|
||||
throws Exception
|
||||
{
|
||||
int max = 8;
|
||||
Exception last = null;
|
||||
for (int attempt = 1; attempt <= max; attempt++)
|
||||
{
|
||||
DefaultTransactionDefinition def = new DefaultTransactionDefinition();
|
||||
def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
|
||||
TransactionStatus tx = transactionManager.getTransaction(def);
|
||||
try
|
||||
{
|
||||
product.setEffectiveDate(forcedEff);
|
||||
if (importPsvId != null)
|
||||
{
|
||||
product.setPsvId(importPsvId);
|
||||
}
|
||||
int[] nums = importOneProduct(workbook, product, operName, regionByNameCn, euRegions,
|
||||
zoneByCountryAndCode, unmatchedCountries, forcedEff);
|
||||
transactionManager.commit(tx);
|
||||
return nums;
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
try { transactionManager.rollback(tx); } catch (Exception ignore) { }
|
||||
last = e;
|
||||
if (attempt >= max || !isImportDeadlock(e))
|
||||
{
|
||||
throw e;
|
||||
}
|
||||
log.warn("product import deadlock, code={}, retry {}/{}", product.getProductCode(),
|
||||
Integer.valueOf(attempt), Integer.valueOf(max));
|
||||
try
|
||||
{
|
||||
Thread.sleep(80L * attempt + (long) (Math.random() * 40));
|
||||
}
|
||||
catch (InterruptedException ie)
|
||||
{
|
||||
Thread.currentThread().interrupt();
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
}
|
||||
throw last != null ? last : new ServiceException("product import retry exhausted");
|
||||
}
|
||||
|
||||
private static boolean isImportDeadlock(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
|
||||
|| name.indexOf("DeadlockLoser") >= 0
|
||||
|| name.indexOf("CannotAcquireLock") >= 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
String msg = cur.getMessage();
|
||||
if (msg != null && (msg.indexOf("Deadlock found when trying to get lock") >= 0
|
||||
|| msg.indexOf("try restarting transaction") >= 0))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
cur = cur.getCause();
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private String resolveCountryLabel(String countryCode)
|
||||
{
|
||||
if (StringUtils.isEmpty(countryCode))
|
||||
|
||||
+25
-1
@@ -996,6 +996,7 @@ public class BmsRedlineProductServiceImpl implements IBmsRedlineProductService
|
||||
d.setIsDel(BmsRedlineProduct.IS_DEL_NO);
|
||||
toInsert.add(d);
|
||||
}
|
||||
sortRedlineCountriesForInsert(toInsert);
|
||||
insertRedlineCountryChunks(toInsert);
|
||||
return toInsert.size();
|
||||
}
|
||||
@@ -1014,13 +1015,36 @@ public class BmsRedlineProductServiceImpl implements IBmsRedlineProductService
|
||||
return taskId;
|
||||
}
|
||||
|
||||
/** Stable insert order reduces InnoDB secondary-index lock inversion. */
|
||||
private static void sortRedlineCountriesForInsert(List<BmsRedlineProductCountry> rows)
|
||||
{
|
||||
if (rows == null || rows.size() < 2)
|
||||
{
|
||||
return;
|
||||
}
|
||||
rows.sort((a, b) -> {
|
||||
int c = String.valueOf(a == null ? null : a.getCountryCode())
|
||||
.compareToIgnoreCase(String.valueOf(b == null ? null : b.getCountryCode()));
|
||||
if (c != 0) return c;
|
||||
c = String.valueOf(a == null ? null : a.getZoneCode())
|
||||
.compareToIgnoreCase(String.valueOf(b == null ? null : b.getZoneCode()));
|
||||
if (c != 0) return c;
|
||||
java.math.BigDecimal aw = a == null ? null : a.getWeightMin();
|
||||
java.math.BigDecimal bw = b == null ? null : b.getWeightMin();
|
||||
if (aw == null && bw == null) return 0;
|
||||
if (aw == null) return -1;
|
||||
if (bw == null) return 1;
|
||||
return aw.compareTo(bw);
|
||||
});
|
||||
}
|
||||
|
||||
private void insertRedlineCountryChunks(List<BmsRedlineProductCountry> rows)
|
||||
{
|
||||
if (rows == null || rows.isEmpty())
|
||||
{
|
||||
return;
|
||||
}
|
||||
int size = 200;
|
||||
int size = 100;
|
||||
for (int i = 0; i < rows.size(); i += size)
|
||||
{
|
||||
int end = Math.min(i + size, rows.size());
|
||||
|
||||
+23
-19
@@ -1,36 +1,38 @@
|
||||
package com.bms.price.service.impl;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.FileOutputStream;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.util.Collections;
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import org.apache.commons.io.FilenameUtils;
|
||||
import org.apache.commons.io.IOUtils;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Service;
|
||||
import com.bms.common.config.BmsConfig;
|
||||
import com.bms.common.exception.ServiceException;
|
||||
import com.bms.common.utils.DateUtils;
|
||||
import com.bms.common.utils.StringUtils;
|
||||
import com.bms.common.utils.file.FileUploadUtils;
|
||||
import com.bms.common.utils.file.OssStorage;
|
||||
import com.bms.common.utils.uuid.IdUtils;
|
||||
import com.bms.price.domain.PrcPriceExcelFile;
|
||||
import com.bms.price.mapper.PrcPriceExcelFileMapper;
|
||||
import com.bms.price.service.IPrcPriceExcelFileService;
|
||||
|
||||
/**
|
||||
* Persist import Excel bytes and metadata
|
||||
* Persist import Excel bytes and metadata (OSS, same pattern as AR bill export).
|
||||
*/
|
||||
@Service
|
||||
public class PrcPriceExcelFileServiceImpl implements IPrcPriceExcelFileService
|
||||
{
|
||||
private static final String SUB_DIR = "price/excel";
|
||||
/** Object key prefix under configured OSS pathPrefix. */
|
||||
private static final String OBJECT_DIR = "price/excel";
|
||||
|
||||
@Autowired
|
||||
private PrcPriceExcelFileMapper excelFileMapper;
|
||||
|
||||
@Autowired
|
||||
private OssStorage ossStorage;
|
||||
|
||||
@Override
|
||||
public PrcPriceExcelFile selectPrcPriceExcelFileById(Long pefId)
|
||||
{
|
||||
@@ -71,30 +73,32 @@ public class PrcPriceExcelFileServiceImpl implements IPrcPriceExcelFileService
|
||||
ext = "xlsx";
|
||||
}
|
||||
ext = ext.toLowerCase(Locale.ROOT);
|
||||
String type = excelType.trim().toUpperCase(Locale.ROOT);
|
||||
String contentType = resolveContentType(ext);
|
||||
|
||||
String uploadDir = BmsConfig.getUploadPath() + File.separator + SUB_DIR;
|
||||
String pathName = DateUtils.datePath() + "/" + IdUtils.fastSimpleUUID() + "." + ext;
|
||||
File dest = FileUploadUtils.getAbsoluteFile(uploadDir, pathName);
|
||||
FileOutputStream fos = null;
|
||||
String objectKey = OBJECT_DIR + "/" + type.toLowerCase(Locale.ROOT) + "/"
|
||||
+ DateUtils.datePath() + "/" + IdUtils.fastSimpleUUID() + "." + ext;
|
||||
Path temp = Files.createTempFile("price-excel-", "." + ext);
|
||||
String storageKey;
|
||||
try
|
||||
{
|
||||
fos = new FileOutputStream(dest);
|
||||
fos.write(bytes);
|
||||
Files.write(temp, bytes);
|
||||
storageKey = ossStorage.upload(temp, objectKey, contentType);
|
||||
}
|
||||
finally
|
||||
{
|
||||
IOUtils.close(fos);
|
||||
try { Files.deleteIfExists(temp); } catch (Exception ignore) { }
|
||||
}
|
||||
String storedPath = FileUploadUtils.getPathFileName(uploadDir, pathName);
|
||||
|
||||
PrcPriceExcelFile row = new PrcPriceExcelFile();
|
||||
row.setExcelType(excelType.trim().toUpperCase(Locale.ROOT));
|
||||
row.setExcelType(type);
|
||||
row.setPsvId(psvId);
|
||||
row.setEffectiveTime(effectiveTime);
|
||||
row.setFileName(originalName);
|
||||
row.setFilePath(storedPath);
|
||||
// Persist oss://bucket/key like AR bill storageKey (download via /common/download/resource).
|
||||
row.setFilePath(storageKey);
|
||||
row.setFileSize(Long.valueOf(bytes.length));
|
||||
row.setContentType(resolveContentType(ext));
|
||||
row.setContentType(contentType);
|
||||
row.setFileExtension(ext);
|
||||
row.setCreateBy(operName);
|
||||
excelFileMapper.insertPrcPriceExcelFile(row);
|
||||
@@ -124,6 +128,6 @@ public class PrcPriceExcelFileServiceImpl implements IPrcPriceExcelFileService
|
||||
{
|
||||
return "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
|
||||
}
|
||||
return "";
|
||||
return "application/octet-stream";
|
||||
}
|
||||
}
|
||||
|
||||
+23
-39
@@ -7,11 +7,7 @@ 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;
|
||||
@@ -145,11 +141,13 @@ public class BmsProductImportAsyncRunner
|
||||
return merged;
|
||||
}
|
||||
|
||||
/** Save preview products in parallel. Each product is its own transaction. */
|
||||
/**
|
||||
* Save preview products sequentially (same psv_id).
|
||||
* Parallel inserts into prc_price_set_detail deadlock on secondary indexes.
|
||||
*/
|
||||
@Async("threadPoolTaskExecutor")
|
||||
public void runApplyPreview(String taskId, CatalogPreviewApplyRequest request)
|
||||
{
|
||||
ExecutorService pool = null;
|
||||
try
|
||||
{
|
||||
Map<String, BmsProduct> meta = new HashMap<String, BmsProduct>();
|
||||
@@ -176,6 +174,8 @@ public class BmsProductImportAsyncRunner
|
||||
}
|
||||
}
|
||||
}
|
||||
// Stable order reduces lock-order inversion if another session writes the same version.
|
||||
jobs.sort((a, b) -> String.valueOf(a.getKey()).compareToIgnoreCase(String.valueOf(b.getKey())));
|
||||
int n = jobs.size();
|
||||
if (n == 0)
|
||||
{
|
||||
@@ -184,36 +184,22 @@ public class BmsProductImportAsyncRunner
|
||||
}
|
||||
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)
|
||||
for (int i = 0; i < n; i++)
|
||||
{
|
||||
Map.Entry<String, List<BmsProductCountry>> e = jobs.get(i);
|
||||
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()
|
||||
applyWithDeadlockRetry(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);
|
||||
productService.applyOneCatalogProduct(psvId, code, product, details);
|
||||
}
|
||||
}));
|
||||
}
|
||||
for (Future<?> future : futures)
|
||||
{
|
||||
future.get();
|
||||
});
|
||||
put(taskId, BmsProductImportTask.STATUS_RUNNING, 5 + (90 * (i + 1) / n),
|
||||
"\u6b63\u5728\u4fdd\u5b58 " + (i + 1) + "/" + n, null);
|
||||
}
|
||||
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));
|
||||
@@ -229,19 +215,12 @@ public class BmsProductImportAsyncRunner
|
||||
}
|
||||
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;
|
||||
int max = 8;
|
||||
for (int attempt = 1; ; attempt++)
|
||||
{
|
||||
try
|
||||
@@ -258,7 +237,7 @@ public class BmsProductImportAsyncRunner
|
||||
log.warn("product catalog save deadlock, retry {}/{}", Integer.valueOf(attempt), Integer.valueOf(max));
|
||||
try
|
||||
{
|
||||
Thread.sleep(50L * attempt);
|
||||
Thread.sleep(80L * attempt + (long) (Math.random() * 40));
|
||||
}
|
||||
catch (InterruptedException ie)
|
||||
{
|
||||
@@ -274,17 +253,22 @@ public class BmsProductImportAsyncRunner
|
||||
Throwable cur = ex;
|
||||
while (cur != null)
|
||||
{
|
||||
if (cur instanceof org.springframework.dao.PessimisticLockingFailureException)
|
||||
if (cur instanceof org.springframework.dao.PessimisticLockingFailureException
|
||||
|| cur instanceof org.springframework.dao.CannotAcquireLockException
|
||||
|| cur instanceof org.springframework.dao.DeadlockLoserDataAccessException)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
String name = cur.getClass().getName();
|
||||
if (name.indexOf("MySQLTransactionRollbackException") >= 0)
|
||||
if (name.indexOf("MySQLTransactionRollbackException") >= 0
|
||||
|| name.indexOf("DeadlockLoser") >= 0
|
||||
|| name.indexOf("CannotAcquireLock") >= 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
String msg = cur.getMessage();
|
||||
if (msg != null && msg.indexOf("Deadlock found when trying to get lock") >= 0)
|
||||
if (msg != null && (msg.indexOf("Deadlock found when trying to get lock") >= 0
|
||||
|| msg.indexOf("try restarting transaction") >= 0))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
+22
-39
@@ -7,11 +7,7 @@ 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;
|
||||
@@ -145,11 +141,13 @@ public class BmsRedlineProductImportAsyncRunner
|
||||
return merged;
|
||||
}
|
||||
|
||||
/** Save preview products in parallel. Each product is its own transaction. */
|
||||
/**
|
||||
* Save preview products sequentially (same psv_id).
|
||||
* Parallel inserts into prc_price_set_detail deadlock on secondary indexes.
|
||||
*/
|
||||
@Async("threadPoolTaskExecutor")
|
||||
public void runApplyPreview(String taskId, RedlineCatalogPreviewApplyRequest request)
|
||||
{
|
||||
ExecutorService pool = null;
|
||||
try
|
||||
{
|
||||
Map<String, BmsRedlineProduct> meta = new HashMap<String, BmsRedlineProduct>();
|
||||
@@ -176,6 +174,7 @@ public class BmsRedlineProductImportAsyncRunner
|
||||
}
|
||||
}
|
||||
}
|
||||
jobs.sort((a, b) -> String.valueOf(a.getKey()).compareToIgnoreCase(String.valueOf(b.getKey())));
|
||||
int n = jobs.size();
|
||||
if (n == 0)
|
||||
{
|
||||
@@ -184,36 +183,22 @@ public class BmsRedlineProductImportAsyncRunner
|
||||
}
|
||||
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)
|
||||
for (int i = 0; i < n; i++)
|
||||
{
|
||||
Map.Entry<String, List<BmsRedlineProductCountry>> e = jobs.get(i);
|
||||
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()
|
||||
applyWithDeadlockRetry(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);
|
||||
productService.applyOneCatalogProduct(psvId, code, product, details);
|
||||
}
|
||||
}));
|
||||
}
|
||||
for (Future<?> future : futures)
|
||||
{
|
||||
future.get();
|
||||
});
|
||||
put(taskId, BmsRedlineProductImportTask.STATUS_RUNNING, 5 + (90 * (i + 1) / n),
|
||||
"\u6b63\u5728\u4fdd\u5b58 " + (i + 1) + "/" + n, null);
|
||||
}
|
||||
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));
|
||||
@@ -229,19 +214,12 @@ public class BmsRedlineProductImportAsyncRunner
|
||||
}
|
||||
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;
|
||||
int max = 8;
|
||||
for (int attempt = 1; ; attempt++)
|
||||
{
|
||||
try
|
||||
@@ -258,7 +236,7 @@ public class BmsRedlineProductImportAsyncRunner
|
||||
log.warn("redline catalog save deadlock, retry {}/{}", Integer.valueOf(attempt), Integer.valueOf(max));
|
||||
try
|
||||
{
|
||||
Thread.sleep(50L * attempt);
|
||||
Thread.sleep(80L * attempt + (long) (Math.random() * 40));
|
||||
}
|
||||
catch (InterruptedException ie)
|
||||
{
|
||||
@@ -274,17 +252,22 @@ public class BmsRedlineProductImportAsyncRunner
|
||||
Throwable cur = ex;
|
||||
while (cur != null)
|
||||
{
|
||||
if (cur instanceof org.springframework.dao.PessimisticLockingFailureException)
|
||||
if (cur instanceof org.springframework.dao.PessimisticLockingFailureException
|
||||
|| cur instanceof org.springframework.dao.CannotAcquireLockException
|
||||
|| cur instanceof org.springframework.dao.DeadlockLoserDataAccessException)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
String name = cur.getClass().getName();
|
||||
if (name.indexOf("MySQLTransactionRollbackException") >= 0)
|
||||
if (name.indexOf("MySQLTransactionRollbackException") >= 0
|
||||
|| name.indexOf("DeadlockLoser") >= 0
|
||||
|| name.indexOf("CannotAcquireLock") >= 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
String msg = cur.getMessage();
|
||||
if (msg != null && msg.indexOf("Deadlock found when trying to get lock") >= 0)
|
||||
if (msg != null && (msg.indexOf("Deadlock found when trying to get lock") >= 0
|
||||
|| msg.indexOf("try restarting transaction") >= 0))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user