第五章限时订单实战笔记

古城微笑少年丶 2023-05-30 06:30 155阅读 0赞

什么是限时订单？在各种电商网站下订单后会保留一个时间段，时间段内未支付则自动将订单状态设置为已过期，这种订单称之为限时订单。

代码地址：[https://gitee.com/hankin\_chj/rocketmq-platform.git][https_gitee.com_hankin_chj_rocketmq-platform.git] （[rocket-delay-order][]）

# **一、如何实现限时订单** #

## **1、限时订单的流程** ##

电商平台都会包含以下 5 种状态。

待付款：代表买家下单了但是还没有付款。

待发货：代表买家付款了卖家还没有发货。

已发货：代表卖家已经发货并寄出商品了。

已完成：代表买家已经确认收到货了。

已关闭：代表订单过期了买家也没付款、或者卖家关闭了订单。

## **2、限时订单实现的关键** ##

我们可以看到，订单中的很多状态都是可以用户触发的，唯独订单过期了买家也没付款我们需要自动的把订单给关闭，这个操作是没有用户或者是人工干预的，所以限时订单的关键就是如何检查订单状态，如果订单过期了则把该订单设置为关闭状态。

## **3、轮询数据库？** ##

轮询数据库在实现限时订单上是可行的，而且实现起来很简单，写个定时器去每隔一段时间扫描数据库，检查到订单过期了，做适当的业务处理。

但是轮询会带来什么问题？

1）轮询大部分时间其实是在做无用功，我们假设一张订单是45分钟过期，每1分钟我们扫描一次，对这张订单来说，要扫描45次以后，才会检查到这张订单过期，这就意味着数据库的资源（连接，IO）被白白浪费了；

2）处理上的不及时，一个待支付的电影票订单我们假设是12:00:35过期，但是上次扫描的时间是 12:00:30，那么这个订单实际的过期时间是12:01:30，和我本来的过期时间差了55秒钟。放在业务上，会带来什么问题？这张电影票，假设是最后一张，有个人12:00:55来买票，买得到吗？当然买不到了。那么这张电影票很有可能就浪费了。如果缩短扫描的时间间隔，第一只能改善不能解决，第二，又会对数据库造成更大的压力。 那么我们能否有种机制，不用定时扫描，当订单到期了，自然通知我们的应用去处理这些到期的订单呢？

## **4、Java本身的提供的解决方案** ##

java其实已经为我们提供了解决问题的方法。我们想要处理限时支付的问题，肯定是要有个地方保存这些限时订单的信息的，意味着我们需要一个容器，于是我们在Java容器中去寻找Map? List? Queue?

看看java为我们提供的容器，我们是个多线程下的应用，会有多个用户同时下订单，所以所有并发不安全的容器首先被排除，并发安全的容器有哪些？java在阻塞队列里为我们提供了一种叫延迟队列delayQueue的容器，刚好可以为我们解决问题。

**DelayQueue****：阻塞队列（先进先出）**

1）支持阻塞的插入方法：意思是当队列满时，队列会阻塞插入元素的线程，直到队列不满。

2）支持阻塞的移除方法：意思是在队列为空时，获取元素的线程会等待队列变为非空。延迟期满时才能从中提取元素（光队列里有元素还不行）。

Delayed接口使对象成为延迟对象，它使存放在DelayQueue类中的对象具有了激活日期，该接口强制实现下列两个方法：

• CompareTo(Delayed o)：Delayed接口继承了Comparable接口，因此有了这个方法，让元素按激活日期排队。

• getDelay(TimeUnit unit):这个方法返回到激活日期的剩余时间，时间单位由单位参数指定。 阻塞队列更多详情，参考《并发编程》。

## **5、架构师应该多考虑一点** ##

架构师在设计和实现系统时需要考虑些什么？

**功能：**这个没什么好说，实现一个应用，连基本的功能都没实现，要这个应用有何用？

**高性能：**能不能尽快的为用户提供服务和能为多少用户同时提供服务，性能这个东西是个很综合性的东西，从前端到后端，从架构（缓存机制、异步机制）到 web 容器、数据库本身再到虚拟机到算法、java 代码、sql语句的编写，全部都对性能有影响。如何提升性能，要建立在充分的性能测试的基础上，然后一个个的去解决性能瓶颈。对上面提到的应用来讲，我们不想去轮询数据库，其实跟性能有非常大的关系。

**高可用：**应用正确处理业务，服务用户的时间，这个时间当然是越长越好，希望可以7\*24小时。而且哪怕服务器出现了升级，宕机等等情况下，能够以最短的时间恢复，为用户继续服务，但是实际过程中没有哪个网站可以说做到100%，不管是Google、FaceBook、阿里、腾讯，一般来说可以做到99.99%的可用性，已经是相当厉害了，这个水平大概就是一个服务在一年可以做到只有50分钟不可用。这个需要技术、资金、技术人员的水平和责任心，还要运气。

**高伸缩：**伸缩性是指通过不断向集群中加入服务器的手段来缓解不断上升的用户并发访问压力和不断增长的数据存储需求。就像弹簧一样挂东西一样，用户多，伸一点，用户少，缩一点。衡量架构是否高伸缩性的主要标准就是是否可用多台服务器构建集群，是否容易向集群中添加新的服务器。加入新的服务器后是否可以提供和原来服务器无差别的服务。集群中可容纳的总的服务器数量是否有限制。

**高扩展：**的主要标准就是在网站增加新的业务产品时，是否可以实现对现有产品透明无影响，不需要任何改动或者很少改动既有业务功能就可以上线新产品。比如购买电影票的应用，用户购买电影票，现在我们要增加一个功能，用户买了票后，随机抽取用户送限量周边。怎么做到不改动用户下订单功能的基础上增加这个功能。熟悉设计模式的同学，应该很眼熟，这是设计模式中的开闭原则（对扩展开放，对修改关闭）在架构层面的一个原则。

## **6、从系统可用性角度考虑** ##

**应用重启带来的问题**：

保存在Queue中的订单会丢失，这些丢失的订单会在什么时候过期，因为队列里已经没有这个订单了，无法检查了，这些订单就得不到处理了。

已过期的订单不会被处理，在应用的重启阶段，可能会有一部分订单过期，这部分过期未支付的订单同样也得不到处理，会一直放在数据库里，过期未支付订单所对应的资源比如电影票所对应的座位，就不能被释放出来，让别的用户来购买。

解决之道 ：在系统启动时另行处理

## **7、从系统伸缩性角度考虑** ##

集群化了会带来什么问题？应用之间会相互抢夺订单，特别是在应用重启的时候，重新启动的那个应用会把不属于自己的订单，也全部加载到自己的队列里去，一是造成内存的浪费，二来会造成订单的重复处理，而且加大了数据库的压力。

解决方案：让应用分区处理

1）给每台服务器编号，然后在订单表里登记每条订单的服务器编号；

2）更简单的，在订单表里登记每台服务器的IP地址，修改相应的sql语句即可。

几个问题：如果有一台服务器挂了怎么办？如果是某台服务器下线或者宕机，起不来怎么搞？这个还是还是稍微有点麻烦，需要人工干预一下，手动把库里的每条订单数据的服务器编号改为目前正常的服务器的编号，不过也就是一条sql语句的事，然后想办法让正常的服务器进行处理（重启正常的服务器）。

# **二、用RocketMQ实现限时订单** #

引入RocketMQ使用延时消息，一举解决我们限时订单的伸缩性和扩展性问题。

## **1、延时消息**  ##

**概念介绍**

**延时消息：**Producer将消息发送到消息队列RocketMQ服务端，但并不期望这条消息立马投递，而是延迟一定时间后才投递到Consumer进行消费，该消息即延时消息。

**适用场景**

**消息生产和消费有时间窗口要求：**比如在电商交易中超时未支付关闭订单的场景，在订单创建时会发送一条延时消息。这条消息将会在30分钟以后投递给消费者，消费者收到此消息后需要判断对应的订单是否已完成支付；如支付未完成，则关闭订单，如已完成支付则忽略。

## **2、核心的代码** ##

整个代码见delayOrder包，Git地址：https://gitee.com/hankin\_chj/rocketmq-platform.git

### **2.1、配置部分** ###

**  
<**bean** **id****="rocketMQProducer"** **class****="com.chj.service.mq.RocketMQProducer"**  
**     ****init-method****="init"** **destroy-method****="destroy"**>  
   <**property** **name****="producerGroup"** **value****="DelayOrderProducer"** />  
   <**property** **name****="namesrvAddr"** **value****="127.0.0.1:9876"** />  
</**bean**>  
**  
<**bean** **id****="messageListeners"** **class****="com.chj.service.mq.MessageListenerImpl"**></**bean**>  
**  
<**bean** **id****="rocketmqConsumer"** **class****="org.apache.rocketmq.client.consumer.DefaultMQPushConsumer"**  
**     ****init-method****="start"** **destroy-method****="shutdown"**>  
   <**property** **name****="consumerGroup"** **value****="TimeOrderGroup"** />  
   <**property** **name****="namesrvAddr"** **value****="127.0.0.1:9876"** />  
   <**property** **name****="messageModel"** **value****="CLUSTERING"** />  
   <**property** **name****="consumeFromWhere"** **value****="CONSUME\_FROM\_LAST\_OFFSET"** />  
   <**property** **name****="messageListener"** **ref****="messageListeners"** />  
   <**property** **name****="subscription"**>  
      <**map**>  
         <**entry** **key****="TimeOrder"** **value****="\*"** />  
      </**map**>  
   </**property**>  
</**bean**>

订单处理的控制器代码实现：

@Controller
    public class OrderController {
        private static final String SUCCESS = "suc";
        private static final String FAILUER = "failure";
    
        @Autowired
        private SaveOrder saveOrder;
    
        @RequestMapping("/index")
        public String userOrder(){
            return "order";
        }
        //保存订单(界面生成几个订单)
        @RequestMapping("/submitOrder")
        @ResponseBody
        public String saveOrder(@RequestParam("orderNumber")int orderNumber){
           saveOrder.insertOrders(orderNumber);
           return SUCCESS;
        }
    }

### **2.2、核心代码实现** ###

1）保存订单SaveOrder.java的时候，作为生产者往消息队列里推入订单，核心RocketMQProducer，这个类当然是要继承IDelayOrder，同时也是RocketMQ的生产者。

订单相关的服务SaveOrder.java代码实现：

@Service
    public class SaveOrder {
       private Logger logger = LoggerFactory.getLogger(SaveOrder.class);
       public final static short UNPAY = 0;
       public final static short PAYED = 1;
       public final static short EXPIRED = -1;
       @Autowired
       private OrderExpDao orderExpDao;
       @Autowired
       @Qualifier("rocketmq")
       private IDelayOrder delayOrder;
       /**
        * 接收前端页面参数，生成订单
        * @param orderNumber 订单个数
        */
        public void insertOrders(int orderNumber){
           Random r = new Random();
           OrderExp orderExp ;
           for(int i=0;i<orderNumber;i++) {
             //这个是设置延时消息的属性
             //"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"  18个等级
             long expire_duration =30;
             long expireTime =4;
                orderExp = new OrderExp();
                String orderNo = "DD00_30S";
                orderExp.setOrderNo(orderNo);
                orderExp.setOrderNote("享学订单——"+orderNo);
                orderExp.setOrderStatus(UNPAY);
                orderExpDao.insertDelayOrder(orderExp,expire_duration);
                logger.info("保存订单到DB:"+orderNo);
                //TODO 这里需要把订单信息存入RocketMQ
                delayOrder.orderDelay(orderExp, expireTime);
           }
        }
        @PostConstruct
        public void initDelayOrder() {
           logger.info("系统启动，扫描表中过期未支付的订单并处理.........");
           int counts = orderExpDao.updateExpireOrders();
           logger.info("系统启动，处理了表中["+counts+"]个过期未支付的订单！");
           List<OrderExp> orderList = orderExpDao.selectUnPayOrders();
           logger.info("系统启动，发现了表中还有["+orderList.size()+"]个未到期未支付的订单！推入检查队列准备到期检查....");
           for(OrderExp order:orderList) {
              long expireTime = order.getExpireTime().getTime()-(new Date().getTime());
              delayOrder.orderDelay(order, expireTime);
           }
        }
    }

消息队列的实现RocketMQProducer：

@Service
    @Qualifier("rocketmq")
    public class RocketMQProducer implements IDelayOrder {
        @Autowired
        private DlyOrderProcessor processDelayOrder;
        private Thread takeOrder;
        private static final Logger logger = LoggerFactory.getLogger(RocketMQProducer.class);
        private DefaultMQProducer defaultMQProducer;
        private String producerGroup;
        private String namesrvAddr;
    
        @PostConstruct
        public void init() throws MQClientException {
            this.defaultMQProducer = new DefaultMQProducer(this.producerGroup);
            defaultMQProducer.setNamesrvAddr(this.namesrvAddr);
            defaultMQProducer.start();
            logger.info("rocketMQ初始化生产者完成[producerGroup：" + producerGroup + "]");
        }
        @PreDestroy
        public void destroy() {
            defaultMQProducer.shutdown();
            logger.info("rocketMQ生产者[producerGroup: " + producerGroup + "]已停止");
        }
        public DefaultMQProducer getDefaultMQProducer() {
            return defaultMQProducer;
        }
        public void setProducerGroup(String producerGroup) {
            this.producerGroup = producerGroup;
        }
        public void setNamesrvAddr(String namesrvAddr) {
            this.namesrvAddr = namesrvAddr;
        }
        public void orderDelay(OrderExp order, long timeLevel) {
            try {
                //TODO 使用Gson序列化
                Gson gson = new Gson();
                String txtMsg = gson.toJson(order);
                //TODO 发送延时消息
                Message msg = new Message("TimeOrder", null, txtMsg.getBytes());
                //这个是设置延时消息的属性
                //"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"  18个等级
                msg.setDelayTimeLevel((int)timeLevel);
                SendResult result = defaultMQProducer.send(msg);
                if(result.getSendStatus() !=null && result.getSendStatus()== SendStatus.SEND_OK){
                    System.out.println("订单被推入延迟队列，订单详情:"+order);
                    logger.info("订单被推入延迟队列，订单详情："+order);
                }else{
                    logger.error("订单推入RocketMq失败，订单详情："+order+"SendStatus:"+result.getSendStatus());
                }
            } catch (Exception e) {
                logger.error("单推入RocketMq失败，失败详情："+e.toString());
            }
        }
    }

2）消息队列会把延时的订单发给消费者MessageListenerImpl，它是一个RocketMQ的消费者监听，它来负责检查订单是否过期，有消息过来，证明消息订单过期了，则把订单状态修改为过期订单。RocketMQ本身又如何保证可用性和伸缩性？这个就需要RocketMQ的主从同步(HA机制)。

处理消息队列返回的延时订单MessageListenerImpl：

@Service
    public class MessageListenerImpl implements MessageListenerConcurrently {
        private Logger logger = LoggerFactory.getLogger(MessageListenerImpl.class);
        @Autowired
        private DlyOrderProcessor processDlyOrder;
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            for (MessageExt msg : msgs) {
                try {
                    //TODO 使用GSON反序列化
                    String txtMsg = new String(msg.getBody());
                    Gson gson = new Gson();
                    System.out.println("接收到RocketMQ的消息："+txtMsg);
                    OrderExp order = (OrderExp)gson.fromJson(txtMsg, OrderExp.class);
                    //TODO 修改订单状态为过期
                    if(order.getId()!=null){
                        processDlyOrder.checkDelayOrder(order);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
            // 如果没有异常会认为都成功消费
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }

处理过期订单的服务：

@Service
    public class DlyOrderProcessor {
       private Logger logger = LoggerFactory.getLogger(DlyOrderProcessor.class);
       @Autowired
       private OrderExpDao orderExpDao;
       /**
        * 检查数据库中指定id的订单的状态,如果为未支付，则修改为已过期
        * */
       public void checkDelayOrder(OrderExp record) {
          OrderExp dbOrder = orderExpDao.selectByPrimaryKey(record.getId());
          if(dbOrder.getOrderStatus()==SaveOrder.UNPAY) {
             logger.info("订单【"+record+"】未支付已过期，需要更改为过期订单！");
             orderExpDao.updateExpireOrder(record.getId());
          }else {
             logger.info("已支付订单【"+record+"】，无需修改！");
          }
       }
    }

[https_gitee.com_hankin_chj_rocketmq-platform.git]: https://gitee.com/hankin_chj/rocketmq-platform.git
[rocket-delay-order]: https://gitee.com/hankin_chj/rocketmq-platform/tree/master/rocket-delay-order