redis延迟队列,处理正常订单超时自动关闭

梦里梦外; 2024-04-17 17:07 67阅读 0赞

demo代码地址

链接：https://pan.baidu.com/s/1yY\_84ANvwh78gE6G0HXnBg   
提取码：gzvu

<dependency>
                <groupId>org.redisson</groupId>
                <artifactId>redisson</artifactId>
                <version>3.9.1</version>
            </dependency>

1.创建redis连接 连接池

import org.redisson.Redisson;
    import org.redisson.api.RedissonClient;
    import org.redisson.config.ClusterServersConfig;
    import org.redisson.config.Config;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * @author : zhang sq
     * @date : 2019/8/22 11:56
     **/
    @Configuration
    public class RedisConfig {
    
        @Value("${redis.addresses}")
        private String[] addresses;  //地址配置在配置文件上
    
        @Bean
        public RedissonClient createRedisAPi(){
            //redis集群配置 start
            Config redissonConfig = new Config();
            //改用redisson后为了之间数据能兼容，这里修改编码为org.redisson.client.codec.StringCodec
            redissonConfig.setCodec(new org.redisson.client.codec.StringCodec());
            ClusterServersConfig clusterServersConfig = redissonConfig.useClusterServers();
            clusterServersConfig.setScanInterval(2000).addNodeAddress(addresses);
    //                .addNodeAddress("redis://127.0.0.1:7000")
    //                .addNodeAddress("redis://127.0.0.1:7001")
    //                .addNodeAddress("redis://127.0.0.1:7002")
    //                .addNodeAddress("redis://127.0.0.1:7003")
    //                .addNodeAddress("redis://127.0.0.1:7004")
    //                .addNodeAddress("redis://127.0.0.1:7005");
            //设置对于master节点的连接池中连接数最大为500
            clusterServersConfig.setMasterConnectionPoolSize(500);
            //设置密码
    //        clusterServersConfig.setPassword("1qaz@WSX");
            //设置对于slave节点的连接池中连接数最大为500
            clusterServersConfig.setSlaveConnectionPoolSize(500);
            //如果当前连接池里的连接数量超过了最小空闲连接数，而同时有连接空闲时间超过了该数值，
            // 那么这些连接将会自动被关闭，并从连接池里去掉。时间单位是毫秒。
            clusterServersConfig.setIdleConnectionTimeout(10000);
            //同任何节点建立连接时的等待超时。时间单位是毫秒。
            clusterServersConfig.setConnectTimeout(30000);
            //等待节点回复命令的时间。该时间从命令发送成功时开始计时。
            clusterServersConfig.setTimeout(3000);
            clusterServersConfig.setPingTimeout(30000);
            //当与某个节点的连接断开时，等待与其重新建立连接的时间间隔。时间单位是毫秒。
            clusterServersConfig.setReconnectionTimeout(3000);
            //redis集群配置 end
            return Redisson.create(redissonConfig);
    
            //单redis连接 start
    //        SingleServerConfig singleServerConfig = redissonConfig.useSingleServer();
    //        singleServerConfig.setAddress("redis://127.0.0.1:6379");
    //        singleServerConfig.setPassword("zsq2170");
    //          设置redis几号数据库
    //        singleServerConfig.setDatabase(0);
    //        singleServerConfig.setClientName("jiuyue");
    //        singleServerConfig.setConnectTimeout(10000);
    //        singleServerConfig.setConnectionPoolSize(300);
    //        return Redisson.create(redissonConfig);
            //单redis连接 end
        }
    }

2.编写测试 生产者 ,正式生产应该在提交订单的时候生产

@RestController
    @RequestMapping("/test")
    @Api(tags = "生产者",description = "OrderProducerController")
    public class OrderProducerController {
    
        @Autowired
        private RedissonClient redisson; //注入redis连接
    
        @PostMapping("/producer")
        @ApiOperation(value = "生产")
            public void Producer(){
            RBlockingDeque<Order> blockingDeque = redisson.getBlockingDeque("jiuyang_order_message"); //获取阻塞队列<名称和消费时获取名称一致>
            RDelayedQueue<Order> delayedQueue = redisson.getDelayedQueue(blockingDeque); //加入延迟队列
            for (long i = 100; i < 120; i++) {  //循环放20个
                try {
                    Order order=new Order();
                    order.setId(123L);
                    //orderInfo.setCreateTime(DateHelper.date2String(new Date()));
                    delayedQueue.offer(order,20, TimeUnit.SECONDS);  //将对象发送到redis延迟队列 ,20 单位秒 <这里单位可以选择的>
                    System.out.println("===="+i);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

3.编写消费者消费 创建配置类,启动服务时便执行消费方法

@Component
    public class ApplicationRunnerImpl implements ApplicationRunner {
    
        @Autowired
        private RefundOrderServiceImpl refundOrderService;
        
        //tomcat启动执行此方法
        @Override
        public void run(ApplicationArguments args) throws Exception {
            refundOrderService.refundOrder();
        }
    }

4.多线程处理业务,先创建线程池

@Component
    public class MyAsyncConfigurer implements AsyncConfigurer {
    
        private static final Logger log = LoggerFactory.getLogger(MyAsyncConfigurer.class);
    
        @Override
        @Bean
        public ThreadPoolTaskExecutor getAsyncExecutor() {
            ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();
            //设置核心线程数，默认为1
            threadPool.setCorePoolSize(4);
            // 当核心线程都在跑任务，还有多余的任务会存到此处。
            threadPool.setQueueCapacity(30);
            //最大线程数，默认为Integer.MAX_VALUE，如果queueCapacity存满了，还有任务就会启动更多的线程，直到线程数达到maxPoolSize。如果还有任务，则根据拒绝策略进行处理。
            threadPool.setMaxPoolSize(8);
            // 设置线程活跃时间（秒）
            threadPool.setKeepAliveSeconds(120);
            threadPool.setWaitForTasksToCompleteOnShutdown(true);
            threadPool.setAwaitTerminationSeconds(60 * 15);
            threadPool.setThreadNamePrefix("MyAsync-");
            threadPool.initialize();
            return threadPool;
        }
    }

5.多线程消费延迟队列任务

@Service
    @Transactional
    public class RefundOrderServiceImpl implements RefundOrderService {
    
        private static final Logger logger = LoggerFactory.getLogger(RefundOrderServiceImpl.class);
    
        @Autowired
        private RefundOrderMapper refundOrderMapper;
    
        @Autowired
        private RedissonClient redissonClient;  //redis
    
        @Autowired
        private ThreadPoolTaskExecutor scheduledThreadPoolExecutor; //线程
    
        //读取阻塞队列,多线程执行
        public void refundOrder() {
            RBlockingDeque<Order> blockingDeque = redissonClient.getBlockingDeque("jiuyang_order_message");
            scheduledThreadPoolExecutor.execute(new SendingTask(blockingDeque));
        }
    
    
        private class SendingTask implements Runnable {
    
            RBlockingDeque<Order> blockingDeque = null;
    
            public SendingTask(RBlockingDeque<Order> blockingDeque) {
                this.blockingDeque = blockingDeque;
            }
    
            @Override
            public void run() {
                while (true) {
                    try {
                        /**
                         * blockqueue,当拿到数据时，才往下执行.没有则等待
                         */
                        Order order = blockingDeque.take();
                        scheduledThreadPoolExecutor.execute(new BusiTask(order));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    
    
        private class BusiTask implements Runnable {
    
            Order order = null;
    
            public BusiTask(Order order) {
                this.order = order;
            }
    
            @Override
            @Transactional
            public void run() {
                try {
                    if (order != null) {
                        Long orderId = order.getId();
                        String orderSn = order.getOrderSn();
                        //查询订单状态
                        Order newOrder = refundOrderMapper.selectOrderById(orderId, orderSn);
                        //订单状态：0->待付款；1->待发货；2->已发货；3->已收货待评价；4->已关闭；5->无效订单；6->已评价；8->退款中
                        Integer status = newOrder.getStatus();
                        if (status == 0) {
                            refundOrderMapper.updateStatus(orderId, orderSn, 4); //关闭订单
                            logger.info("未支付订单已关闭");
                            System.err.println("未支付订单已关闭");
                            //退还商品库存  查询订单商品表退库存
                           List<OmsOrderItem> orderItemList= refundOrderMapper.selectQuantity(orderId,orderSn);
                           if(TargetUtils.listNotNull(orderItemList)){
                               for (OmsOrderItem orderItem:orderItemList){
                                   String productId = orderItem.getProductId(); //商品id
                                   Integer quantity = orderItem.getProductQuantity(); //购买数量
                                   //修改库存,退款库存
                                   refundOrderMapper.updateProductSku(productId,quantity);
                                   logger.info("未支付订单购买商品库存已退还");
                                   System.err.println("未支付订单购买商品库存已退还");
                               }
                           }
                           //退还优惠券
                            Long couponId = newOrder.getCouponId(); //获取优惠券id
                            if (null != couponId && couponId > 0) { //有优惠券
                                refundOrderMapper.updateCouponId(couponId); //退还优惠券
                                logger.info("优惠券已退还");
                                System.err.println("优惠券已退还");
                            }
                            //退还礼品卡钱
                            String cardSn = newOrder.getCardSn();//卡号
                            if(StringUtils.isNotEmpty(cardSn)){
                                BigDecimal cardAmount = newOrder.getCardAmount(); //消费金额
                                //退还礼品卡消费金额
                                refundOrderMapper.updateCardOrder(cardSn,cardAmount);
                                logger.info("礼品卡消费金额已退还");
                                System.err.println("礼品卡消费金额已退还");
                            }
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    
    
    }