backend/java-mall
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化 seata 的死锁释放配置

Browse Source
This commit is contained in:
Jack 2025-08-28 23:20:58 +08:00
parent 425aebf601
commit 6c29c0d6cc
6 changed files with 267 additions and 242 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
mall-account/src/main/resources/bootstrap-prod.yml
View File

@ -86,6 +86,12 @@ seata:
service: service:
vgroup-mapping: vgroup-mapping:
my_test_tx_group: default my_test_tx_group: default
client:
rm:
lock:
lock-timeout: 60000
retry-times: 5
retry-interval-milliseconds: 2000
# 控制台日志信息 # 控制台日志信息
logging: logging:
level: level:

7
mall-pay/src/main/resources/bootstrap-prod.yml
View File

@ -27,6 +27,7 @@ spring:
aop-patterns: com.suisung.mall.common.service.impl.*,com.suisung.mall.core.web.service.impl.*,com.suisung.mall.pay.* aop-patterns: com.suisung.mall.common.service.impl.*,com.suisung.mall.core.web.service.impl.*,com.suisung.mall.pay.*
remove-abandoned: true remove-abandoned: true
remove-abandoned-timeout: 1800 remove-abandoned-timeout: 1800
transaction-query-timeout: 3000
redis: redis:
host: @redis.host@ # Redis服务器地址 host: @redis.host@ # Redis服务器地址
database: @redis.database@ # Redis数据库索引（默认为0） database: @redis.database@ # Redis数据库索引（默认为0）
@ -86,6 +87,12 @@ seata:
service: service:
vgroup-mapping: vgroup-mapping:
my_test_tx_group: default my_test_tx_group: default
client:
rm:
lock:
lock-timeout: 60000
retry-times: 5
retry-interval-milliseconds: 2000
# 控制台日志信息 # 控制台日志信息
logging: logging:
level: level:

133
mall-shop/src/main/java/com/suisung/mall/shop/sync/listen/ShopBatchSubmitListener.java
View File

@ -21,7 +21,9 @@ import org.springframework.stereotype.Component;
import java.math.BigDecimal; import java.math.BigDecimal;
import java.util.*; import java.util.*;
import java.util.concurrent.*; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -29,42 +31,35 @@ import java.util.stream.Collectors;
@Slf4j @Slf4j
public class ShopBatchSubmitListener extends AnalysisEventListener<SxGoosModelExcel> { public class ShopBatchSubmitListener extends AnalysisEventListener<SxGoosModelExcel> {
// 批处理阈值 // 批处理阈值
private static final int BATCH_SIZE = 500; private static final int BATCH_SIZE = 490;
// 数据缓存
private List<SxGoosModelExcel> cachedDataList = new ArrayList<>(BATCH_SIZE);
private SyncThirdDataService syncThirdDataService;
private SyncStoreSpecsService syncStoreSpecsService;
// 线程池配置 // 线程池配置
private final ExecutorService executorService; private final ExecutorService executorService;
// 数据缓存
private List<Future<?>> futures ; private final List<SxGoosModelExcel> cachedDataList = new ArrayList<>(BATCH_SIZE);
private AtomicInteger success; private final SyncThirdDataService syncThirdDataService;
private AtomicInteger fails; private final SyncStoreSpecsService syncStoreSpecsService;
private AtomicInteger batchSize; private final List<Future<?>> futures;
private final AtomicInteger success;
private final AtomicInteger fails;
private final AtomicInteger batchSize;
private final ProductSpecManager productSpecManager;
@Setter @Setter
@Getter @Getter
private String storeId; private String storeId;
@Setter @Setter
@Getter @Getter
private String isNegativeAllowed; private String isNegativeAllowed;
@Setter @Setter
@Getter @Getter
private Map<String, Integer> brandMaps; private Map<String, Integer> brandMaps;
private ProductSpecManager productSpecManager;
public ShopBatchSubmitListener(SyncThirdDataService syncThirdDataService, SyncStoreSpecsService syncStoreSpecsService) { public ShopBatchSubmitListener(SyncThirdDataService syncThirdDataService, SyncStoreSpecsService syncStoreSpecsService) {
this.syncThirdDataService = syncThirdDataService; this.syncThirdDataService = syncThirdDataService;
this.syncStoreSpecsService = syncStoreSpecsService; this.syncStoreSpecsService = syncStoreSpecsService;
// 创建线程池（根据CPU核心数优化） // 创建线程池（根据CPU核心数优化）
int corePoolSize = Runtime.getRuntime().availableProcessors(); int corePoolSize = Runtime.getRuntime().availableProcessors();
log.info("核心线程数量{}" , corePoolSize); // log.info("核心线程数量{}", corePoolSize);
this.executorService = Executors.newFixedThreadPool(6); this.executorService = Executors.newFixedThreadPool(corePoolSize);
this.futures = new ArrayList<>(); this.futures = new ArrayList<>();
this.success = new AtomicInteger(); this.success = new AtomicInteger();
this.fails = new AtomicInteger(); this.fails = new AtomicInteger();
@ -74,6 +69,54 @@ public class ShopBatchSubmitListener extends AnalysisEventListener<SxGoosModelEx
productSpecManager.setSpecsMap(specsMap); productSpecManager.setSpecsMap(specsMap);
} }
public static void main(String[] args) {
SxGoosModelExcel sxGoosModelExcel = new SxGoosModelExcel();
sxGoosModelExcel.setProduct_barcode("1");
sxGoosModelExcel.setProduct_name("产品1");
sxGoosModelExcel.setShop_specs("颜色");
sxGoosModelExcel.setRetail_price(new BigDecimal("100"));
sxGoosModelExcel.setStock(new BigDecimal("10"));
sxGoosModelExcel.setJsonSpecs("[颜色:红色][尺寸:10]");
SxGoosModelExcel sxGoosModelExcel1 = new SxGoosModelExcel();
sxGoosModelExcel1.setProduct_barcode("2");
sxGoosModelExcel1.setProduct_name("产品2");
sxGoosModelExcel1.setShop_specs("颜色");
sxGoosModelExcel1.setShop_spec("红色");
sxGoosModelExcel1.setRetail_price(new BigDecimal("200"));
sxGoosModelExcel1.setStock(new BigDecimal("20"));
sxGoosModelExcel1.setJsonSpecs("[颜色:黑色][尺寸:20]");
SxGoosModelExcel sxGoosModelExcel3 = new SxGoosModelExcel();
sxGoosModelExcel3.setProduct_barcode("1");
sxGoosModelExcel3.setProduct_name("产品3");
sxGoosModelExcel3.setShop_specs("颜色");
sxGoosModelExcel3.setShop_spec("黑色");
sxGoosModelExcel3.setRetail_price(new BigDecimal("300"));
sxGoosModelExcel3.setStock(new BigDecimal("30"));
sxGoosModelExcel3.setJsonSpecs("[颜色:黑色][尺寸:30]");
List<SxGoosModelExcel> sxGoosModelExcelList = new ArrayList<>();
sxGoosModelExcelList.add(sxGoosModelExcel);
sxGoosModelExcelList.add(sxGoosModelExcel1);
sxGoosModelExcelList.add(sxGoosModelExcel3);
ProductSpecManager productSpecManager = new ProductSpecManager();
Map<String, List<String>> specsMap = new HashMap<>();
productSpecManager.setSpecsMap(specsMap);
sxGoosModelExcelList = sxGoosModelExcelList.stream().peek(gm -> {
if (StringUtils.isNotEmpty(gm.getJsonSpecs())) {
List<ProductSpecManager.SpecItem> specItems = productSpecManager.parseSpecString(gm.getJsonSpecs());
productSpecManager.addProductSpec(gm.getProduct_barcode(), specItems, gm.getRetail_price(), gm.getStock());
}
}).filter(CommonUtil.distinctByKey(SxGoosModelExcel::getProduct_barcode))
.collect(Collectors.toList());
Gson goson = new Gson();
log.info("规格:{}", productSpecManager.getSepcMapValue("1"));
log.info("规格数量:{}", productSpecManager.getSepcMapValue("1").size());
log.info(String.valueOf(sxGoosModelExcelList.size()));
}
@Override @Override
public void invoke(SxGoosModelExcel sxGoosModelExcel, AnalysisContext analysisContext) { public void invoke(SxGoosModelExcel sxGoosModelExcel, AnalysisContext analysisContext) {
synchronized (cachedDataList) { synchronized (cachedDataList) {
@ -146,6 +189,7 @@ public class ShopBatchSubmitListener extends AnalysisEventListener<SxGoosModelEx
/** /**
* 数据处理 * 数据处理
*
* @param list * @param list
* @return * @return
*/ */
@ -218,6 +262,7 @@ public class ShopBatchSubmitListener extends AnalysisEventListener<SxGoosModelEx
/** /**
* 查找存量数据 * 查找存量数据
*
* @param listMap * @param listMap
* @return * @return
*/ */
@ -238,52 +283,4 @@ public class ShopBatchSubmitListener extends AnalysisEventListener<SxGoosModelEx
return resultMap; return resultMap;
} }
public static void main(String[] args) {
SxGoosModelExcel sxGoosModelExcel = new SxGoosModelExcel();
sxGoosModelExcel.setProduct_barcode("1");
sxGoosModelExcel.setProduct_name("产品1");
sxGoosModelExcel.setShop_specs("颜色");
sxGoosModelExcel.setRetail_price(new BigDecimal("100"));
sxGoosModelExcel.setStock(new BigDecimal("10"));
sxGoosModelExcel.setJsonSpecs("[颜色:红色][尺寸:10]");
SxGoosModelExcel sxGoosModelExcel1 = new SxGoosModelExcel();
sxGoosModelExcel1.setProduct_barcode("2");
sxGoosModelExcel1.setProduct_name("产品2");
sxGoosModelExcel1.setShop_specs("颜色");
sxGoosModelExcel1.setShop_spec("红色");
sxGoosModelExcel1.setRetail_price(new BigDecimal("200"));
sxGoosModelExcel1.setStock(new BigDecimal("20"));
sxGoosModelExcel1.setJsonSpecs("[颜色:黑色][尺寸:20]");
SxGoosModelExcel sxGoosModelExcel3 = new SxGoosModelExcel();
sxGoosModelExcel3.setProduct_barcode("1");
sxGoosModelExcel3.setProduct_name("产品3");
sxGoosModelExcel3.setShop_specs("颜色");
sxGoosModelExcel3.setShop_spec("黑色");
sxGoosModelExcel3.setRetail_price(new BigDecimal("300"));
sxGoosModelExcel3.setStock(new BigDecimal("30"));
sxGoosModelExcel3.setJsonSpecs("[颜色:黑色][尺寸:30]");
List<SxGoosModelExcel> sxGoosModelExcelList = new ArrayList<>();
sxGoosModelExcelList.add(sxGoosModelExcel);
sxGoosModelExcelList.add(sxGoosModelExcel1);
sxGoosModelExcelList.add(sxGoosModelExcel3);
ProductSpecManager productSpecManager=new ProductSpecManager();
Map<String, List<String>> specsMap=new HashMap<>();
productSpecManager.setSpecsMap(specsMap);
sxGoosModelExcelList=sxGoosModelExcelList.stream().peek(gm -> {
if(StringUtils.isNotEmpty(gm.getJsonSpecs())){
List<ProductSpecManager.SpecItem> specItems= productSpecManager.parseSpecString(gm.getJsonSpecs());
productSpecManager.addProductSpec(gm.getProduct_barcode(),specItems,gm.getRetail_price(),gm.getStock());
}
}).filter(CommonUtil.distinctByKey(SxGoosModelExcel::getProduct_barcode))
.collect(Collectors.toList());
Gson goson=new Gson();
log.info("规格:{}",productSpecManager.getSepcMapValue("1"));
log.info("规格数量:{}",productSpecManager.getSepcMapValue("1").size());
log.info(String.valueOf(sxGoosModelExcelList.size()));
}
} }

82
mall-shop/src/main/java/com/suisung/mall/shop/sync/pool/ThreadPoolManager.java
View File

@ -31,11 +31,14 @@ public class ThreadPoolManager {
private volatile boolean isShutdown = false; private volatile boolean isShutdown = false;
public ThreadPoolManager(int corePoolSize, int maxPoolSize, String successFile, String failureFile) { public ThreadPoolManager(int corePoolSize, int maxPoolSize, String successFile, String failureFile) {
this.successFile = successFile; this.successFile = successFile;
this.failureFile = failureFile; this.failureFile = failureFile;
// 动态分配
corePoolSize = Math.max(4, Runtime.getRuntime().availableProcessors() * 2);
maxPoolSize = corePoolSize * 2;
this.executor = new ThreadPoolExecutor( this.executor = new ThreadPoolExecutor(
corePoolSize, corePoolSize,
maxPoolSize, maxPoolSize,
@ -54,41 +57,6 @@ public class ThreadPoolManager {
Runtime.getRuntime().availableProcessors() * 2, Runtime.getRuntime().availableProcessors() * 2,
successFile, failureFile); successFile, failureFile);
} }
// 自定义线程工厂
private static class CustomThreadFactory implements ThreadFactory {
private final AtomicInteger counter = new AtomicInteger(0);
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, "pool-thread-" + counter.incrementAndGet());
thread.setPriority(Thread.NORM_PRIORITY);
thread.setDaemon(false);
return thread;
}
}
// 自定义拒绝策略
private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
if (!isShutdown) {
// 1. 尝试直接执行（调用者线程）
try {
System.out.println("线程池饱和，由调用线程直接执行");
r.run();
} catch (Exception e) {
System.err.println("调用者执行任务失败: " + e.getMessage());
}
// 2. 记录指标
failureCount.incrementAndGet();
} else {
System.err.println("线程池已关闭，拒绝新任务");
}
}
}
// 消费待处理任务的线程 // 消费待处理任务的线程
private void startPendingTaskConsumer() { private void startPendingTaskConsumer() {
@ -110,7 +78,6 @@ public class ThreadPoolManager {
consumerThread.start(); consumerThread.start();
} }
/** /**
* 提交文件夹处理任务（带重试机制） * 提交文件夹处理任务（带重试机制）
*/ */
@ -136,7 +103,7 @@ public class ThreadPoolManager {
System.out.printf("[%s] 任务处理失败，第%d次重试...%n", System.out.printf("[%s] 任务处理失败，第%d次重试...%n",
taskName, retryCount); taskName, retryCount);
try { try {
Thread.sleep(1000 * retryCount); // 指数退避 Thread.sleep(1000L * retryCount); // 指数退避
} catch (InterruptedException ie) { } catch (InterruptedException ie) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();
break; break;
@ -155,6 +122,7 @@ public class ThreadPoolManager {
/** /**
* 批量提交文件夹处理任务 * 批量提交文件夹处理任务
*
* @param taskName 任务名称(用于日志记录) * @param taskName 任务名称(用于日志记录)
* @param folderPaths 要处理的文件夹路径列表 * @param folderPaths 要处理的文件夹路径列表
* @param callback 处理完成的回调接口 * @param callback 处理完成的回调接口
@ -287,6 +255,7 @@ public class ThreadPoolManager {
monitorThread.setDaemon(true); monitorThread.setDaemon(true);
monitorThread.start(); monitorThread.start();
} }
private void printPoolStatus() { private void printPoolStatus() {
System.out.println("\n=== 线程池状态 ==="); System.out.println("\n=== 线程池状态 ===");
System.out.println("活跃线程: " + executor.getActiveCount()); System.out.println("活跃线程: " + executor.getActiveCount());
@ -326,7 +295,7 @@ public class ThreadPoolManager {
// 3. 尝试恢复 - 中断死锁线程 // 3. 尝试恢复 - 中断死锁线程
if (executor instanceof ThreadPoolExecutor) { if (executor instanceof ThreadPoolExecutor) {
ThreadPoolExecutor tpe = (ThreadPoolExecutor) executor; ThreadPoolExecutor tpe = executor;
BlockingQueue<Runnable> queue = tpe.getQueue(); BlockingQueue<Runnable> queue = tpe.getQueue();
System.err.println("尝试中断死锁线程并清空队列(" + queue.size() + "个任务)"); System.err.println("尝试中断死锁线程并清空队列(" + queue.size() + "个任务)");
@ -374,6 +343,7 @@ public class ThreadPoolManager {
executor.setMaximumPoolSize(executor.getMaximumPoolSize() + 5); executor.setMaximumPoolSize(executor.getMaximumPoolSize() + 5);
} }
} }
/** /**
* 关闭线程池（优雅关闭） * 关闭线程池（优雅关闭）
*/ */
@ -411,4 +381,38 @@ public class ThreadPoolManager {
public interface JsonProcessCallback { public interface JsonProcessCallback {
void process(JSONArray jsonArray, String taskName, String fileName); void process(JSONArray jsonArray, String taskName, String fileName);
} }
// 自定义线程工厂
private static class CustomThreadFactory implements ThreadFactory {
private final AtomicInteger counter = new AtomicInteger(0);
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, "pool-thread-" + counter.incrementAndGet());
thread.setPriority(Thread.NORM_PRIORITY);
thread.setDaemon(false);
return thread;
}
}
// 自定义拒绝策略
private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
if (!isShutdown) {
// 1. 尝试直接执行（调用者线程）
try {
System.out.println("线程池饱和，由调用线程直接执行");
r.run();
} catch (Exception e) {
System.err.println("调用者执行任务失败: " + e.getMessage());
}
// 2. 记录指标
failureCount.incrementAndGet();
} else {
System.err.println("线程池已关闭，拒绝新任务");
}
}
}
} }

7
mall-shop/src/main/resources/bootstrap-prod.yml
View File

@ -33,6 +33,7 @@ spring:
aop-patterns: com.suisung.mall.common.service.impl.*,com.suisung.mall.core.web.service.impl.*,com.suisung.mall.shop.* aop-patterns: com.suisung.mall.common.service.impl.*,com.suisung.mall.core.web.service.impl.*,com.suisung.mall.shop.*
remove-abandoned: true remove-abandoned: true
remove-abandoned-timeout: 1800 remove-abandoned-timeout: 1800
transaction-query-timeout: 3000
redis: redis:
host: @redis.host@ # Redis服务器地址 host: @redis.host@ # Redis服务器地址
database: @redis.database@ # Redis数据库索引（默认为0） database: @redis.database@ # Redis数据库索引（默认为0）
@ -100,6 +101,12 @@ seata:
service: service:
vgroup-mapping: vgroup-mapping:
my_test_tx_group: default my_test_tx_group: default
client:
rm:
lock:
lock-timeout: 60000
retry-times: 5
retry-interval-milliseconds: 2000
# 控制台日志信息 # 控制台日志信息
logging: logging:
level: level:

6
pom.xml
View File

@ -317,6 +317,7 @@
<!-- seata配置 --> <!-- seata配置 -->
<seata.group>SEATA_GROUP</seata.group> <seata.group>SEATA_GROUP</seata.group>
<seata.tx-service-group>my_test_tx_group</seata.tx-service-group> <seata.tx-service-group>my_test_tx_group</seata.tx-service-group>
<seata.server.address>114.132.210.208:8091</seata.server.address>
<!-- sentinel配置 --> <!-- sentinel配置 -->
<sentinel.transport.dashboard>114.132.210.208:8718</sentinel.transport.dashboard> <sentinel.transport.dashboard>114.132.210.208:8718</sentinel.transport.dashboard>
<!-- mysql配置 --> <!-- mysql配置 -->
@ -367,6 +368,7 @@
<!-- seata配置 --> <!-- seata配置 -->
<seata.group>SEATA_GROUP</seata.group> <seata.group>SEATA_GROUP</seata.group>
<seata.tx-service-group>my_test_tx_group</seata.tx-service-group> <seata.tx-service-group>my_test_tx_group</seata.tx-service-group>
<seata.server.address>114.132.210.208:8091</seata.server.address>
<!-- sentinel配置 --> <!-- sentinel配置 -->
<sentinel.transport.dashboard>114.132.210.208:8718</sentinel.transport.dashboard> <sentinel.transport.dashboard>114.132.210.208:8718</sentinel.transport.dashboard>
<!-- mysql配置 --> <!-- mysql配置 -->
@ -417,6 +419,7 @@
<!-- seata配置 --> <!-- seata配置 -->
<seata.group>SEATA_GROUP</seata.group> <seata.group>SEATA_GROUP</seata.group>
<seata.tx-service-group>my_test_tx_group</seata.tx-service-group> <seata.tx-service-group>my_test_tx_group</seata.tx-service-group>
<seata.server.address>114.132.210.208:8091</seata.server.address>
<!-- sentinel配置 --> <!-- sentinel配置 -->
<sentinel.transport.dashboard>10.1.8.3:8718</sentinel.transport.dashboard> <sentinel.transport.dashboard>10.1.8.3:8718</sentinel.transport.dashboard>
<!-- mysql配置 --> <!-- mysql配置 -->
@ -467,6 +470,7 @@
<!-- seata配置 --> <!-- seata配置 -->
<seata.group>SEATA_GROUP</seata.group> <seata.group>SEATA_GROUP</seata.group>
<seata.tx-service-group>my_test_tx_group</seata.tx-service-group> <seata.tx-service-group>my_test_tx_group</seata.tx-service-group>
<!-- <seata.server.address>172.16.0.11:8091</seata.server.address>-->
<!-- sentinel配置 --> <!-- sentinel配置 -->
<sentinel.transport.dashboard>172.16.0.11:8718</sentinel.transport.dashboard> <sentinel.transport.dashboard>172.16.0.11:8718</sentinel.transport.dashboard>
<!-- mysql配置 --> <!-- mysql配置 -->
@ -550,7 +554,7 @@
<!-- <baseImage>openjdk:8-jre-alpine</baseImage>--> <!-- <baseImage>openjdk:8-jre-alpine</baseImage>-->
<baseImage>openjdk:8-jre</baseImage> <baseImage>openjdk:8-jre</baseImage>
<!--定义容器启动命令，注意不能换行--> <!--定义容器启动命令，注意不能换行-->
<entryPoint>["java", "-jar", "-Xms256m", "-Xmx512m", "-XX:MetaspaceSize=256m", "-XX:MaxMetaspaceSize=256m", "-XX:+UseContainerSupport", "-XX:MaxRAMPercentage=60.0", "-XX:+UseSerialGC", "-XX:MinHeapFreeRatio=40", "-XX:MaxHeapFreeRatio=60", "-Dspring.profiles.active=${spring.profile}", "-Duser.timezone=Asia/Shanghai", "/${project.build.finalName}.jar"] <entryPoint>["java", "-jar", "-Xms256m", "-Xmx512m", "-XX:MetaspaceSize=256m", "-XX:MaxMetaspaceSize=256m", "-XX:+UseContainerSupport", "-XX:MaxRAMPercentage=60.0", "-XX:+UseSerialGC", "-XX:MinHeapFreeRatio=40", "-XX:MaxHeapFreeRatio=60", "-XX:+PrintGCDetails", "-XX:+PrintGCDateStamps", "-Xloggc:./gc.log", "-XX:+UseGCLogFileRotation", "-XX:NumberOfGCLogFiles=5", "-XX:GCLogFileSize=10M", "-Dspring.profiles.active=${spring.profile}", "-Duser.timezone=Asia/Shanghai", "/${project.build.finalName}.jar"]
</entryPoint> </entryPoint>
<!--推送镜像仓库校验安全证书，无安全证书无法推送--> <!--推送镜像仓库校验安全证书，无安全证书无法推送-->