From 8ce196d1a317177b8546fc787e8c28cd97a78eec Mon Sep 17 00:00:00 2001 From: xmzhao71 Date: Wed, 6 Sep 2023 14:16:18 +0800 Subject: [PATCH] =?UTF-8?q?feat=EF=BC=9A=E6=B7=BB=E5=8A=A0=E8=87=AA?= =?UTF-8?q?=E5=AE=9A=E4=B9=89=E7=BA=BF=E7=A8=8B=E6=B1=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- mallinkService/pom.xml | 6 ++ .../com/iformall/service/WxOrderService.java | 2 +- .../service/helper/WxOrderServiceHelper.java | 16 ++-- .../thread/CompletableFutureWithExecutor.java | 38 +++++++++ .../service/thread/ThreadPoolService.java | 81 +++++++++++++++++++ .../main/resources/mapper/WxOrderMapper.xml | 2 +- 6 files changed, 135 insertions(+), 10 deletions(-) create mode 100644 mallinkService/src/main/java/com/iformall/service/thread/CompletableFutureWithExecutor.java create mode 100644 mallinkService/src/main/java/com/iformall/service/thread/ThreadPoolService.java 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}