diff --git a/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/handle/OrderAutoConfirmListener.java b/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/handle/OrderAutoConfirmListener.java new file mode 100644 index 0000000..d5772b3 --- /dev/null +++ b/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/handle/OrderAutoConfirmListener.java @@ -0,0 +1,36 @@ +package co.yixiang.yshop.module.order.handle; + +import cn.hutool.core.util.ObjectUtil; +import cn.hutool.core.util.StrUtil; +import co.yixiang.yshop.framework.common.constant.ShopConstants; +import co.yixiang.yshop.framework.tenant.core.util.TenantUtils; +import co.yixiang.yshop.module.message.redismq.DelayedQueueListener; +import co.yixiang.yshop.module.message.redismq.msg.OrderMsg; +import co.yixiang.yshop.module.order.service.storeorder.AppStoreOrderService; +import jakarta.annotation.Resource; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +/** + * 自动收货订单监听 + */ +@Component +@Slf4j +public class OrderAutoConfirmListener implements DelayedQueueListener { + @Resource + private AppStoreOrderService appStoreOrderService; + @Override + public String delayedQueueKey() { + return ShopConstants.REDIS_ORDER_OUTTIME_UNCONFIRM; + } + + @Override + public void consume(OrderMsg message) throws Exception { + if(ObjectUtil.isNotNull(message) && StrUtil.isNotEmpty(message.getOrderId())) { + TenantUtils.executeIgnore(() -> { + appStoreOrderService.takeOrder(message.getOrderId(),null); + log.info("订单编号:{}自动确认收货成功",message.getOrderId()); + }); + } + } +} diff --git a/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/handle/OrderUnPayListener.java b/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/handle/OrderUnPayListener.java new file mode 100644 index 0000000..44a2255 --- /dev/null +++ b/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/handle/OrderUnPayListener.java @@ -0,0 +1,36 @@ +package co.yixiang.yshop.module.order.handle; + +import cn.hutool.core.util.ObjectUtil; +import cn.hutool.core.util.StrUtil; +import co.yixiang.yshop.framework.common.constant.ShopConstants; +import co.yixiang.yshop.framework.tenant.core.util.TenantUtils; +import co.yixiang.yshop.module.message.redismq.DelayedQueueListener; +import co.yixiang.yshop.module.message.redismq.msg.OrderMsg; +import co.yixiang.yshop.module.order.service.storeorder.AppStoreOrderService; +import jakarta.annotation.Resource; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +/** + * 未支付订单监听 + */ +@Component +@Slf4j +public class OrderUnPayListener implements DelayedQueueListener { + @Resource + private AppStoreOrderService appStoreOrderService; + @Override + public String delayedQueueKey() { + return ShopConstants.REDIS_ORDER_OUTTIME_UNPAY_QUEUE; + } + + @Override + public void consume(OrderMsg message) throws Exception { + if(ObjectUtil.isNotNull(message) && StrUtil.isNotEmpty(message.getOrderId())) { + TenantUtils.executeIgnore(() -> { + appStoreOrderService.cancelOrder(message.getOrderId(),null); + log.info("订单编号:{}自动取消订单成功",message.getOrderId()); + }); + } + } +} diff --git a/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/handle/RedisDelayHandle.java b/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/handle/RedisDelayHandle.java index 1ffcb9b..39cb738 100644 --- a/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/handle/RedisDelayHandle.java +++ b/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/handle/RedisDelayHandle.java @@ -17,12 +17,13 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.ThreadFactory; /** - * 延时队列消费 + * 延时队列消费 已经被废弃 * @author hupeng * @date 2023.7.27 */ -@Component +//@Component @Slf4j +@Deprecated public class RedisDelayHandle { @Resource private RedissonClient redissonClient; diff --git a/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/service/storeorder/AppStoreOrderServiceImpl.java b/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/service/storeorder/AppStoreOrderServiceImpl.java index aab99bd..8f509b1 100644 --- a/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/service/storeorder/AppStoreOrderServiceImpl.java +++ b/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/service/storeorder/AppStoreOrderServiceImpl.java @@ -23,6 +23,7 @@ import co.yixiang.yshop.module.member.service.useraddress.AppUserAddressService; import co.yixiang.yshop.module.member.service.userbill.UserBillService; import co.yixiang.yshop.module.message.enums.WechatTempateEnum; import co.yixiang.yshop.module.message.mq.producer.WeixinNoticeProducer; +import co.yixiang.yshop.module.message.redismq.msg.OrderMsg; import co.yixiang.yshop.module.order.controller.app.order.param.AppOrderParam; import co.yixiang.yshop.module.order.controller.app.order.param.AppPayParam; import co.yixiang.yshop.module.order.controller.app.order.vo.AppStoreOrderQueryVo; @@ -324,7 +325,7 @@ public class AppStoreOrderServiceImpl extends ServiceImpl blockingDeque = redissonClient.getBlockingDeque(ShopConstants.REDIS_ORDER_OUTTIME_UNPAY_QUEUE ); RDelayedQueue delayedQueue = redissonClient.getDelayedQueue(blockingDeque); - delayedQueue.offer(storeOrder, ShopConstants.ORDER_OUTTIME_UNPAY, TimeUnit.MINUTES); + delayedQueue.offer(OrderMsg.builder().orderId(orderSn).build(), ShopConstants.ORDER_OUTTIME_UNPAY, TimeUnit.MINUTES); String s = TimeUnit.SECONDS.toSeconds(ShopConstants.ORDER_OUTTIME_UNPAY) + "分钟"; log.info("添加延时队列成功 ，延迟时间：" + s + "订单编号：" + orderSn); } catch (Exception e) { diff --git a/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/service/storeorder/StoreOrderServiceImpl.java b/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/service/storeorder/StoreOrderServiceImpl.java index 2db338f..7da4a3b 100644 --- a/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/service/storeorder/StoreOrderServiceImpl.java +++ b/yshop-drink-boot3/yshop-module-mall/yshop-module-order-biz/src/main/java/co/yixiang/yshop/module/order/service/storeorder/StoreOrderServiceImpl.java @@ -18,6 +18,7 @@ import co.yixiang.yshop.module.member.enums.BillDetailEnum; import co.yixiang.yshop.module.member.service.user.MemberUserService; import co.yixiang.yshop.module.member.service.userbill.UserBillService; import co.yixiang.yshop.module.message.mq.producer.WeixinNoticeProducer; +import co.yixiang.yshop.module.message.redismq.msg.OrderMsg; import co.yixiang.yshop.module.order.controller.admin.storeorder.vo.*; import co.yixiang.yshop.module.order.convert.storeorder.StoreOrderConvert; import co.yixiang.yshop.module.order.dal.dataobject.storeorder.StoreOrderDO; @@ -161,7 +162,7 @@ public class StoreOrderServiceImpl implements StoreOrderService { try { RBlockingDeque blockingDeque = redissonClient.getBlockingDeque(ShopConstants.REDIS_ORDER_OUTTIME_UNCONFIRM); RDelayedQueue delayedQueue = redissonClient.getDelayedQueue(blockingDeque); - delayedQueue.offer(updateReqVO.getOrderId(), ShopConstants.ORDER_OUTTIME_UNCONFIRM, TimeUnit.MINUTES); + delayedQueue.offer(OrderMsg.builder().orderId(updateObj.getOrderId()).build(), ShopConstants.ORDER_OUTTIME_UNCONFIRM, TimeUnit.MINUTES); String s = TimeUnit.SECONDS.toSeconds(ShopConstants.ORDER_OUTTIME_UNCONFIRM) + "分钟"; log.info("添加延时队列成功 ，延迟时间：" + s); } catch (Exception e) { diff --git a/yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/DelayedQueueListener.java b/yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/DelayedQueueListener.java new file mode 100644 index 0000000..8f14696 --- /dev/null +++ b/yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/DelayedQueueListener.java @@ -0,0 +1,39 @@ +package co.yixiang.yshop.module.message.redismq; + +/** + * 延时队列接口 + * @author hupeng + * @date 2024.6.25 + */ +public interface DelayedQueueListener { + /** + * 是否启用 + * + * @return boolean + */ + default boolean isEnable() { + return true; + } + + /** + * 队列键 + * + * @return String + */ + String delayedQueueKey(); + + /** + * 消费 + * + * @param message Object + * @throws Exception Exception + */ + void consume(T message) throws Exception; + + /** + * 发生异常时最终处理 + */ + default void whenExceptionFinally() { + } + +} diff --git a/yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/DelayedQueueListenerConfigurer.java b/yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/DelayedQueueListenerConfigurer.java new file mode 100644 index 0000000..e1e55a9 --- /dev/null +++ b/yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/DelayedQueueListenerConfigurer.java @@ -0,0 +1,58 @@ +package co.yixiang.yshop.module.message.redismq; + +import com.google.common.util.concurrent.ThreadFactoryBuilder; +import org.redisson.api.RedissonClient; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.stereotype.Component; +import org.springframework.util.Assert; + +import java.util.List; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * 配置 + * @author hupeng + * @date 2024.6.25 + */ +@Component +public class DelayedQueueListenerConfigurer implements InitializingBean, DisposableBean { + private ThreadPoolExecutor delayedThreadPoolExecutor; + + private final List> delayedQueueListenerList; + + private final RedissonClient redissonClient; + + public DelayedQueueListenerConfigurer(List> delayedQueueListenerList, RedissonClient redissonClient) { + this.delayedQueueListenerList = delayedQueueListenerList; + this.redissonClient = redissonClient; + } + + @Override + public void destroy() throws Exception { + if (delayedThreadPoolExecutor != null) { + delayedThreadPoolExecutor.shutdownNow(); + } + } + + @Override + public void afterPropertiesSet() throws Exception { + Assert.notEmpty(delayedQueueListenerList, "delayedQueueListenerList must not be empty!"); + + ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("delayed-queue-pool-%d").build(); + int numberOfJob = delayedQueueListenerList.size(); + delayedThreadPoolExecutor = new ThreadPoolExecutor( + numberOfJob, + numberOfJob, + 0L, + TimeUnit.MILLISECONDS, + new LinkedBlockingQueue<>(numberOfJob), + namedThreadFactory + ); + delayedQueueListenerList.forEach(delayedQueueListener -> delayedThreadPoolExecutor.execute(new DelayedQueuePollTask<>(redissonClient, delayedQueueListener))); + } + +} diff --git a/yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/DelayedQueuePollTask.java b/yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/DelayedQueuePollTask.java new file mode 100644 index 0000000..029234a --- /dev/null +++ b/yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/DelayedQueuePollTask.java @@ -0,0 +1,47 @@ +package co.yixiang.yshop.module.message.redismq; + +import lombok.extern.slf4j.Slf4j; +import org.redisson.api.RBlockingDeque; +import org.redisson.api.RedissonClient; + +/** + * 延迟队列轮询任务类 + * @author hupeng + * @date 2024.6.25 + */ +@Slf4j +public class DelayedQueuePollTask implements Runnable{ + private final RedissonClient redissonClient; + + private final DelayedQueueListener delayedQueueListener; + + public DelayedQueuePollTask(RedissonClient redissonClient, DelayedQueueListener delayedQueueListener) { + this.redissonClient = redissonClient; + this.delayedQueueListener = delayedQueueListener; + } + + @Override + public void run() { + String threadName = "delayed-queue-listener-" + delayedQueueListener.getClass().getSimpleName(); + Thread.currentThread().setName(threadName); + if (!delayedQueueListener.isEnable()) { + return; + } + RBlockingDeque blockingDeque = redissonClient.getBlockingDeque(delayedQueueListener.delayedQueueKey()); + // 解决消息丢失问题，发送subscribe命令订阅redis队列 + redissonClient.getDelayedQueue(blockingDeque); + while (!Thread.currentThread().isInterrupted()) { + try { + T message = blockingDeque.take(); + delayedQueueListener.consume(message); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (Exception e) { + log.error(e.getMessage(), e); + } finally { + delayedQueueListener.whenExceptionFinally(); + } + } + } + +} diff --git a/yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/msg/OrderMsg.java b/yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/msg/OrderMsg.java new file mode 100644 index 0000000..89dc3b9 --- /dev/null +++ b/yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/msg/OrderMsg.java @@ -0,0 +1,12 @@ +package co.yixiang.yshop.module.message.redismq.msg; + +import lombok.Builder; +import lombok.Data; + +import java.io.Serializable; + +@Data +@Builder +public class OrderMsg implements Serializable { + private String orderId; +}