diff --git a/mallinkService/pom.xml b/mallinkService/pom.xml
index d78dee99c..eafa9ef0e 100644
--- a/mallinkService/pom.xml
+++ b/mallinkService/pom.xml
@@ -40,6 +40,12 @@
alipay-easysdk
2.2.0
+
+
+ com.alibaba
+ transmittable-thread-local
+ 2.11.4
+
\ No newline at end of file
diff --git a/mallinkService/src/main/java/com/iformall/service/WxOrderService.java b/mallinkService/src/main/java/com/iformall/service/WxOrderService.java
index f7a7b781c..325bd8da2 100644
--- a/mallinkService/src/main/java/com/iformall/service/WxOrderService.java
+++ b/mallinkService/src/main/java/com/iformall/service/WxOrderService.java
@@ -198,7 +198,7 @@ public interface WxOrderService {
WxOrderCouponPressVo detailPressVo(WxOrder record);
- void updateStatusByPressCouponId(Long couponId, TenantEntity tenantEntity,int orderStatus);
+ void updateStatusByPressCouponId(Long couponId, TenantEntity tenantEntity, int orderStatus);
/**
* 通过优惠券id更新订单状态
diff --git a/mallinkService/src/main/java/com/iformall/service/helper/WxOrderServiceHelper.java b/mallinkService/src/main/java/com/iformall/service/helper/WxOrderServiceHelper.java
index e2b01ff2f..8ed682216 100644
--- a/mallinkService/src/main/java/com/iformall/service/helper/WxOrderServiceHelper.java
+++ b/mallinkService/src/main/java/com/iformall/service/helper/WxOrderServiceHelper.java
@@ -2,7 +2,6 @@ package com.iformall.service.helper;
import com.iformall.common.ErrorCode;
import com.iformall.domain.po.WxCoupon;
-import com.iformall.domain.po.base.TenantEntity;
import com.iformall.enums.EnumCacheKey;
import com.iformall.enums.EnumOrderStatus;
import com.iformall.enums.EnumYesOrNo;
@@ -10,6 +9,8 @@ import com.iformall.exception.MallinkException;
import com.iformall.service.WxCouponChannelService;
import com.iformall.service.WxCouponService;
import com.iformall.service.WxOrderService;
+import com.iformall.service.thread.CompletableFutureWithExecutor;
+import com.iformall.service.thread.ThreadPoolService;
import com.iformall.utils.RedisLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -17,8 +18,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
-import java.util.concurrent.CompletableFuture;
-
@Service
public class WxOrderServiceHelper {
@@ -39,6 +38,9 @@ public class WxOrderServiceHelper {
@Autowired
RedisLock redisLock;
+ @Autowired
+ private ThreadPoolService threadPoolService;
+
/**
* 微信订单是否可以支付
*/
@@ -69,11 +71,9 @@ public class WxOrderServiceHelper {
}
// 砍价订单,支付时扣库存,库存不足 则修改砍价订单状态
- CompletableFuture.runAsync(() -> {
- if (isPressOrder && null != coupon.getPressPayReduceStock() && coupon.getPressPayReduceStock().intValue() == EnumYesOrNo.YES.getCode().intValue() && stock <= 0 ) {
- wxOrderService.updateStatusByPressCouponId(coupon.getTenantId(), coupon.getCouponId(), EnumOrderStatus.ORDER_STATUS_PRESS_CANCEL.getCode());
- }
- });
+ if (isPressOrder && null != coupon.getPressPayReduceStock() && coupon.getPressPayReduceStock().intValue() == EnumYesOrNo.YES.getCode().intValue() && stock <= 0 ) {
+ CompletableFutureWithExecutor.runAsync(() -> wxOrderService.updateStatusByPressCouponId(coupon.getTenantId(), coupon.getCouponId(), EnumOrderStatus.ORDER_STATUS_PRESS_CANCEL.getCode()));
+ }
}
}
diff --git a/mallinkService/src/main/java/com/iformall/service/thread/CompletableFutureWithExecutor.java b/mallinkService/src/main/java/com/iformall/service/thread/CompletableFutureWithExecutor.java
new file mode 100644
index 000000000..a1b62b445
--- /dev/null
+++ b/mallinkService/src/main/java/com/iformall/service/thread/CompletableFutureWithExecutor.java
@@ -0,0 +1,38 @@
+package com.iformall.service.thread;
+
+import com.alibaba.ttl.threadpool.TtlExecutors;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+
+/**
+ * 使用自定义线程池实现异步编程
+ * @author xmzhao71
+ * @date 2023-09-06
+ */
+@Component
+public class CompletableFutureWithExecutor {
+
+ private static final Integer CORE_SIZE = 2;
+ private static final Integer MAX_SIZE = 2 * Runtime.getRuntime().availableProcessors();
+ private static CompletableFutureWithExecutor future;
+
+ @Autowired
+ private ThreadPoolService threadPoolService;
+
+ @PostConstruct
+ public void init() {
+ future = this;
+ }
+
+ public static CompletableFuture supplyAsync(Supplier supplier) {
+ return CompletableFuture.supplyAsync(supplier, TtlExecutors.getTtlExecutorService(future.threadPoolService.getExecutorService(CORE_SIZE, MAX_SIZE)));
+ }
+
+ public static CompletableFuture runAsync(Runnable runnable) {
+ return CompletableFuture.runAsync(runnable, TtlExecutors.getTtlExecutorService(future.threadPoolService.getExecutorService(CORE_SIZE, MAX_SIZE)));
+ }
+}
diff --git a/mallinkService/src/main/java/com/iformall/service/thread/ThreadPoolService.java b/mallinkService/src/main/java/com/iformall/service/thread/ThreadPoolService.java
new file mode 100644
index 000000000..378557214
--- /dev/null
+++ b/mallinkService/src/main/java/com/iformall/service/thread/ThreadPoolService.java
@@ -0,0 +1,81 @@
+package com.iformall.service.thread;
+
+
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * 自定义线程池
+ * @author xmzhao71
+ * @date 2023-09-06
+ */
+@Component
+public class ThreadPoolService {
+
+ private final Integer coreSize = 2;
+ private final Integer maxSize = 2 * Runtime.getRuntime().availableProcessors();
+ private final String prefix = "fm-thread-pool-";
+
+ private final AtomicInteger nextId = new AtomicInteger();
+
+ /**
+ * 线程池对象
+ */
+ private volatile ThreadPoolExecutor service;
+
+ private ThreadPoolService() {
+ getExecutorService(Integer.MAX_VALUE, Integer.MAX_VALUE);
+ }
+
+ /**
+ * 自定义线程名称,方便的出错的时候溯源
+ */
+ private final ThreadFactory namedThreadFactory = r -> new Thread(r, prefix + nextId.getAndIncrement());
+
+
+ /**
+ * 创建获取线程池对象
+ * corePoolSize 线程池核心池的大小
+ * maximumPoolSize 线程池中允许的最大线程数量
+ * keepAliveTime 当线程数大于核心时,此为终止前多余的空闲线程等待新任务的最长时间
+ * unit keepAliveTime 的时间单位
+ * workQueue 用来储存等待执行任务的队列
+ * threadFactory 创建线程的工厂类
+ * handler 拒绝策略类,当线程池数量达到上线并且workQueue队列长度达到上限时就需要对到来的任务做拒绝处理
+ *
+ * @return 线程池
+ */
+ public synchronized ExecutorService getExecutorService(Integer coreSize, Integer maxSize) {
+
+ if (service == null) {
+ synchronized (ThreadPoolService.class) {
+ if (service == null) {
+ ThreadPoolExecutor executorService = new ThreadPoolExecutor(
+ coreSize,
+ maxSize,
+ 1000L,
+ TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<>(1024),
+ namedThreadFactory,
+ new ThreadPoolExecutor.AbortPolicy()
+ );
+ executorService.allowCoreThreadTimeOut(true);
+ this.service = executorService;
+ }
+ }
+ }
+ return service;
+ }
+
+ /**
+ * 使用线程池创建线程并异步执行任务
+ *
+ * @param r 任务
+ */
+ public void submitTask(Runnable r) {
+ service.execute(r);
+ }
+
+}
diff --git a/mallinkService/src/main/resources/mapper/WxOrderMapper.xml b/mallinkService/src/main/resources/mapper/WxOrderMapper.xml
index a6fd2a1bd..397564255 100644
--- a/mallinkService/src/main/resources/mapper/WxOrderMapper.xml
+++ b/mallinkService/src/main/resources/mapper/WxOrderMapper.xml
@@ -308,7 +308,7 @@
update wx_order set order_status=#{orderStatus}
where product_id=#{productId}
- and press_end_date is not null and order_status in (6,7)
+ and press_end_date is not null and order_status = 6
and `tenant_id` = #{tenantId}