项目实战--订单30分钟自动关闭

订单30分钟自动关闭

一、背景
二、方案分析
- 2.1 数据库轮询
- 2.2 JDK延迟队列
- 2.3 时间轮算法
- 2.4 redis缓存
- 2.5 使用消息队列

一、背景

延时任务的需求：

生成订单30分钟未支付，则自动取消
生成订单60秒后,给用户发短信

二、方案分析

2.1 数据库轮询

该方案通常是在小型项目中使用，即通过一个线程定时的去扫描数据库，通过订单时间来判断是否有超时的订单，然后进行update或delete等操作，早期是用quartz来实现的：

<dependency>
    <groupId>org.quartz-scheduler</groupId>
    <artifactId>quartz</artifactId>
    <version>2.2.2</version>
</dependency>

import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.SchedulerFactory;
import org.quartz.SimpleScheduleBuilder;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.impl.StdSchedulerFactory;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;

public class MyJob implements Job {
    public void execute(JobExecutionContext context)
            throws JobExecutionException {
        System.out.println("要去数据库扫描啦。。。");
    }

    public static void main(String[] args) throws Exception {
        // 创建任务
        JobDetail jobDetail = JobBuilder.newJob(MyJob.class)
                .withIdentity("job1", "group1").build();
        // 创建触发器 每3秒钟执行一次
        Trigger trigger = TriggerBuilder
                .newTrigger()
                .withIdentity("trigger1", "group3")
                .withSchedule(
                        SimpleScheduleBuilder.simpleSchedule()
                                .withIntervalInSeconds(3).repeatForever())
                .build();
        Scheduler scheduler = new StdSchedulerFactory().getScheduler();
        // 将任务及其触发器放入调度器
        scheduler.scheduleJob(jobDetail, trigger);
        // 调度器开始调度任务
        scheduler.start();
    }
}

运行代码，可发现每隔3秒输出。

优点:

简单易行，支持集群操作

缺点:

对服务器内存消耗大
存在延迟，比如你每隔3分钟扫描一次，那最坏的延迟时间就是3分钟
假设订单有几千万条，每隔几分钟这样扫描一次，数据库损耗极大

2.2 JDK延迟队列

该方案是利用JDK自带的DelayQueue来实现，这是一个无界阻塞队列，该队列只有在延迟期满的时候才能从中获取元素，放入DelayQueue中的对象，是必须实现Delayed接口的。
在这里插入图片描述
其中：
poll():获取并移除队列的超时元素，没有则返回空
take():获取并移除队列的超时元素，如果没有则wait当前线程，直到有元素满足超时条件，返回结果。

创建OrderDelay类实现Delayed：

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class OrderDelay implements Delayed {

 private String orderId;
 private long timeout;

 OrderDelay(String orderId, long timeout) {
  this.orderId = orderId;
  this.timeout = timeout + System.nanoTime();
 }

 public int compareTo(Delayed other) {
  if (other == this)
   return 0;
  OrderDelay t = (OrderDelay) other;
  long d = (getDelay(TimeUnit.NANOSECONDS) - t
    .getDelay(TimeUnit.NANOSECONDS));
  return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
 }

 // 返回距离你自定义的超时时间还有多少
 public long getDelay(TimeUnit unit) {
  return unit.convert(timeout - System.nanoTime(), TimeUnit.NANOSECONDS);
 }

 void print() {
  System.out.println(orderId+"编号的订单关闭");
 }
}

设定延迟时间为3秒：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.TimeUnit;

public class DelayQueueDemo {
  public static void main(String[] args) {
         // TODO Auto-generated method stub
         List<String> list = new ArrayList<String>();
         list.add("00000001");
         list.add("00000002");
         list.add("00000003");
         list.add("00000004");
         list.add("00000005");
         DelayQueue<OrderDelay> queue = new DelayQueue<OrderDelay>();
         long start = System.currentTimeMillis();
         for(int i = 0;i<5;i++){
          //延迟三秒取出
             queue.put(new OrderDelay(list.get(i),
                     TimeUnit.NANOSECONDS.convert(3, TimeUnit.SECONDS)));
                 try {
                      queue.take().print();
                      System.out.println("After " +
                              (System.currentTimeMillis()-start) + " MilliSeconds");
             } catch (InterruptedException e) {
                 // TODO Auto-generated catch block
                 e.printStackTrace();
             }
         }
     }
}

优点:

效率高,任务触发时间延迟低。

缺点:

服务器重启后，数据全部消失，怕宕机
集群扩展相当麻烦
因为内存条件限制的原因，比如下单未付款的订单数太多，很容易就出现OOM异常
代码复杂度较高

2.3 时间轮算法

时间轮算法可类比时钟，按某个方向按固定频率轮动，每次跳动称为一个tick。定时轮由3个重要属性参数，ticksPerWheel（一轮的tick数），tickDuration（一个tick的持续时间）以及 timeUnit（时间单位），例如当ticksPerWheel=60，tickDuration=1，timeUnit=秒，这就和现实中的始终的秒针走动完全类似。
在这里插入图片描述
例如上图，当前指针指在1上面，有一个任务需要4秒以后执行，那这个执行的线程回调或者消息将会被放在5上。那如果需要在20秒之后执行，而这个环形结构槽数只到8，如果要20秒，指针需要多转2圈。位置是在2圈之后的5上面（20 % 8 + 1）。
用Netty的HashedWheelTimer来实现：

<dependency>
     <groupId>io.netty</groupId>
     <artifactId>netty-all</artifactId>
     <version>4.1.24.Final</version>
</dependency>

import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.Timer;
import io.netty.util.TimerTask;

import java.util.concurrent.TimeUnit;

public class HashedWheelTimerTest {
 static class MyTimerTask implements TimerTask{
  boolean flag;
  public MyTimerTask(boolean flag){
   this.flag = flag;
  }
  public void run(Timeout timeout) throws Exception {
   // TODO Auto-generated method stub
    System.out.println("要去数据库删除订单了。。。。");
             this.flag =false;
  }
 }
 public static void main(String[] argv) {
  MyTimerTask timerTask = new MyTimerTask(true);
        Timer timer = new HashedWheelTimer();
        timer.newTimeout(timerTask, 5, TimeUnit.SECONDS);
     int i = 1;
        while(timerTask.flag){
         try {
    Thread.sleep(1000);
   } catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
   }
         System.out.println(i+"秒过去了");
         i++;
        }
    }
}

输出:

1秒过去了
2秒过去了
3秒过去了
4秒过去了
5秒过去了
要去数据库删除订单了。。。。
6秒过去了

优点:

效率高,任务触发时间延迟时间比delayQueue低，代码复杂度比delayQueue低。

缺点:

服务器重启后，数据全部消失，怕宕机
集群扩展相当麻烦
因为内存条件限制的原因，比如下单未付款的订单数太多，那么很容易就出现OOM异常

2.4 redis缓存

使用redis的Keyspace Notifications，即键空间机制，利用该机制可以在key失效之后，提供一个回调，实际上是redis会给客户端发送一个消息。是需要redis版本2.8以上。
在redis.conf中，加入一条配置：

notify-keyspace-events Ex

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPubSub;

public class RedisTest {
   private static final String ADDR = "127.0.0.1";
   private static final int PORT = 6379;
   private static JedisPool jedis = new JedisPool(ADDR, PORT);
   private static RedisSub sub = new RedisSub();

 public static void init() {
    new Thread(new Runnable() {
     public void run() {
        jedis.getResource().subscribe(sub, "__keyevent@0__:expired");
     }
    }).start();
 }

 public static void main(String[] args) throws InterruptedException {
    init();
    for(int i =0;i<10;i++){
       String orderId = "OID000000"+i;
       jedis.getResource().setex(orderId, 3, orderId);
       System.out.println(System.currentTimeMillis()+"ms:"+orderId+"订单生成");
    }
 }

 static class RedisSub extends JedisPubSub {
    @Override
    public void onMessage(String channel, String message) {
       System.out.println(System.currentTimeMillis()+"ms:"+message+"订单取消");
    }
   }
}

输出：

1525096202813ms:OID0000000订单生成
1525096202818ms:OID0000001订单生成
1525096202824ms:OID0000002订单生成
1525096202826ms:OID0000003订单生成
1525096202830ms:OID0000004订单生成
1525096202834ms:OID0000005订单生成
1525096202839ms:OID0000006订单生成
1525096205819ms:OID0000000订单取消
1525096205920ms:OID0000005订单取消
1525096205920ms:OID0000004订单取消
1525096205920ms:OID0000001订单取消
1525096205920ms:OID0000003订单取消
1525096205920ms:OID0000006订单取消
1525096205920ms:OID0000002订单取消

Redis的发布/订阅目前是即发即弃(fire and forget)模式的，因此无法实现事件的可靠通知。也就是说，如果发布/订阅的客户端断链之后又重连，则在客户端断链期间的所有事件都会丢失。所以此方案不是太推荐。当然，如果对可靠性要求不高，可以使用。

优点:

由于使用Redis作为消息通道，消息都存储在Redis中。如果发送程序或者任务处理程序挂了，重启之后，还有重新处理数据的可能性。
做集群扩展相当方便
时间准确度高
缺点:
需要额外进行redis维护，数据可能丢失

2.5 使用消息队列

采用rabbitMQ的延时队列。RabbitMQ具有以下两个特性，可以实现延迟队列

RabbitMQ可以针对Queue和Message设置 x-message-tt，来控制消息的生存时间，如果超时，则消息变为dead letter
lRabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key（可选）两个参数，用来控制队列内出现deadletter，则按照这两个参数重新路由。结合两个特性，就可以模拟出延迟消息的功能。

发送MQ的配置：

import net.yjdev.common.core.enums.QueueEnum;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Module 订单支付消息队列
 */
@Configuration
public class RabbitConfig {

    @Value("${spring.rabbitmq.host}")
    private String addresses;

    @Value("${spring.rabbitmq.port}")
    private String port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;


    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(addresses + ":" + port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    /** 订单支付消息队列 */
        @Bean
        public Queue orderpayDirectQueue() {
            return new Queue(QueueEnum.QUEUE_ORDER_PAY.getName(), true, false, false, null);
        }

        @Bean
        public DirectExchange orderpayDirectExchange() {
            return new DirectExchange(QueueEnum.QUEUE_ORDER_PAY.getExchange(), true, false, null);
        }

        @Bean
        public Binding bindingOrderPayExchangeMessage() {
            return BindingBuilder
                    .bind(orderpayDirectQueue())
                    .to(orderpayDirectExchange())
                .with(QueueEnum.QUEUE_ORDER_PAY.getRouteKey());
    }

    /** 订单商品详情消息通知队列 */
    @Bean
    public Queue orderProductlineDirectQueue() {
        return new Queue(QueueEnum.QUEUE_ORDER_PRODUCTLINE.getName(), true, false, false, null);
    }

    @Bean
    public DirectExchange orderProductlineDirectExchange() {
        return new DirectExchange(QueueEnum.QUEUE_ORDER_PRODUCTLINE.getExchange(), true, false, null);
    }

    @Bean
    public Binding bindingOrderProductlineExchangeMessage() {
        return BindingBuilder
                .bind(orderProductlineDirectQueue())
                .to(orderProductlineDirectExchange())
                .with(QueueEnum.QUEUE_ORDER_PRODUCTLINE.getRouteKey());
    }

    @Bean
    public RabbitTemplate newRabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        return template;
    }
}

发送代码：

rabbitTemplate.convertAndSend(QueueEnum.QUEUE_ORDER_DELAY.getExchange(), QueueEnum.QUEUE_ORDER_DELAY.getRouteKey(), orderNo, message -> {
            message.getMessageProperties().setExpiration(Integer.toString(15 * 1000 * 60));
            return message;
        });

订单超时MQ配置：

import net.yjdev.common.core.enums.QueueEnum;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * @Module 订单超时MQ配置
 */

@Configuration
public class DelayRabbitConfig {

    /**
     * 创建一个延时队列
     */
    @Bean
    public Queue delayOrderQueue() {
        Map<String, Object> params = new HashMap<>();
        // x-dead-letter-exchange 声明队列里的死信转发到的DLX名称，
        params.put("x-dead-letter-exchange", QueueEnum.QUEUE_ORDER.getExchange());
        // x-dead-letter-routing-key 声明死信在转发时携带的 routing-key 名称。
        params.put("x-dead-letter-routing-key", QueueEnum.QUEUE_ORDER.getRouteKey());
        return new Queue(QueueEnum.QUEUE_ORDER_DELAY.getName(), true, false, false, params);
    }

    /**
     * 创建一个立即消费队列
     */
    @Bean
    public Queue orderQueue() {
        // 第一个参数为queue的名字，第二个参数为是否支持持久化
        return new Queue(QueueEnum.QUEUE_ORDER.getName(), true);
    }

    /**
     * 延迟交换机
     */
    @Bean
    public DirectExchange orderDelayExchange() {
        // new DirectExchange(ORDER_DELAY_EXCHANGE,true,false);
        return new DirectExchange(QueueEnum.QUEUE_ORDER_DELAY.getExchange(),true,false);
    }

    /**
     * 立即消费交换机
     */
    @Bean
    public TopicExchange orderTopicExchange() {
        return new TopicExchange(QueueEnum.QUEUE_ORDER.getExchange());
    }

    /**
     * 把延时队列和 订单延迟交换的exchange进行绑定
     * @return
     */
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(delayOrderQueue()).to(orderDelayExchange()).with(QueueEnum.QUEUE_ORDER_DELAY.getRouteKey());
    }

    /**
     * 把立即队列和 立即交换的exchange进行绑定
     * @return
     */
    @Bean
    public Binding orderBinding() {
        // 如果要让延迟队列之间有关联,这里的 routingKey 和 绑定的交换机很关键
        return BindingBuilder.bind(orderQueue()).to(orderTopicExchange()).with(QueueEnum.QUEUE_ORDER.getRouteKey());
    }
}

监听延时队列：

@Component
public class OrderCancelMQListener {
	@Autowired
    private IOrderService orderService;
    
	@RabbitListener(queues = "yj.order.cancel")
    @RabbitHandler
    public void orderCancelHandle(SalesOrderCancelVo salesOrderCancelVo, Channel channel, Message message) {
        try {
            log.info("接受到的消息：{}",salesOrderCancelVo.getOrderNo());
            if (StringUtils.isNotEmpty(salesOrderCancelVo.getOrderNo())){
                orderService.cancelSalesOrder(salesOrderCancelVo);
            }else {
                log.info("接受到的消息为空！");
            }
        }catch (Exception e){
            log.error("yj.order.cancel",e);
            try {
                /** 处理消息失败，将消息重新放回队列 */
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }
}