Android中關(guān)于定時任務(wù)實現(xiàn)關(guān)閉訂單問題

更新時間：2022年05月26日 09:06:57 作者：程序員阿牛

在電商、支付等領(lǐng)域，往往會有這樣的場景，用戶下單后放棄支付了，那這筆訂單會在指定的時間段后進行關(guān)閉操作，細心的你一定發(fā)現(xiàn)了像某寶、某東都有這樣的邏輯，而且時間很準確，誤差在1s內(nèi)；那他們是怎么實現(xiàn)的呢？今天通過本文學(xué)習(xí)定時任務(wù)實現(xiàn)關(guān)閉訂單問題

在電商、支付等領(lǐng)域，往往會有這樣的場景，用戶下單后放棄支付了，那這筆訂單會在指定的時間段后進行關(guān)閉操作，細心的你一定發(fā)現(xiàn)了像某寶、某東都有這樣的邏輯，而且時間很準確，誤差在1s內(nèi)；那他們是怎么實現(xiàn)的呢？

一般的做法有如下幾種

定時任務(wù)關(guān)閉訂單
rocketmq延遲隊列
rabbitmq死信隊列
時間輪算法
redis過期監(jiān)聽

一、定時任務(wù)關(guān)閉訂單（最low）

一般情況下，最不推薦的方式就是關(guān)單方式就是定時任務(wù)方式，原因我們可以看下面的圖來說明

我們假設(shè)，關(guān)單時間為下單后10分鐘，定時任務(wù)間隔也是10分鐘；通過上圖我們看出，如果在第1分鐘下單，在第20分鐘的時候才能被掃描到執(zhí)行關(guān)單操作，這樣誤差達到10分鐘，這在很多場景下是不可接受的，另外需要頻繁掃描主訂單號造成網(wǎng)絡(luò)IO和磁盤IO的消耗，對實時交易造成一定的沖擊，所以PASS

二、rocketmq延遲隊列方式

延遲消息 生產(chǎn)者把消息發(fā)送到消息服務(wù)器后，并不希望被立即消費，而是等待指定時間后才可以被消費者消費，這類消息通常被稱為延遲消息。在RocketMQ開源版本中，支持延遲消息，但是不支持任意時間精度的延遲消息，只支持特定級別的延遲消息。消息延遲級別分別為1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，共18個級別。

發(fā)送延遲消息（生產(chǎn)者）

/**
     * 推送延遲消息
     * @param topic 
     * @param body 
     * @param producerGroup 
     * @return boolean
     */
    public boolean sendMessage(String topic, String body, String producerGroup)
    {
        try
        {
            Message recordMsg = new Message(topic, body.getBytes());
            producer.setProducerGroup(producerGroup);
 
            //設(shè)置消息延遲級別，我這里設(shè)置14，對應(yīng)就是延時10分鐘
            // "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
            recordMsg.setDelayTimeLevel(14);
            // 發(fā)送消息到一個Broker
            SendResult sendResult = producer.send(recordMsg);
            // 通過sendResult返回消息是否成功送達
            log.info("發(fā)送延遲消息結(jié)果：======sendResult：{}", sendResult);
            DateFormat format =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            log.info("發(fā)送時間：{}", format.format(new Date()));
 
            return true;
        }
        catch (Exception e)
        {
            e.printStackTrace();
            log.error("延遲消息隊列推送消息異常:{},推送內(nèi)容:{}", e.getMessage(), body);
        }
        return false;
    }

消費延遲消息（消費者）

/**
     * 接收延遲消息
     * 
     * @param topic
     * @param consumerGroup
     * @param messageHandler
     */
    public void messageListener(String topic, String consumerGroup, MessageListenerConcurrently messageHandler)
{
        ThreadPoolUtil.execute(() ->
        {
            try
            {
                DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
                consumer.setConsumerGroup(consumerGroup);
                consumer.setVipChannelEnabled(false);
                consumer.setNamesrvAddr(address);
                //設(shè)置消費者拉取消息的策略，*表示消費該topic下的所有消息，也可以指定tag進行消息過濾
                consumer.subscribe(topic, "*");
                //消費者端啟動消息監(jiān)聽，一旦生產(chǎn)者發(fā)送消息被監(jiān)聽到，就打印消息，和rabbitmq中的handlerDelivery類似
                consumer.registerMessageListener(messageHandler);
                consumer.start();
                log.info("啟動延遲消息隊列監(jiān)聽成功:" + topic);
            }
            catch (MQClientException e)
            {
                log.error("啟動延遲消息隊列監(jiān)聽失敗:{}", e.getErrorMessage());
                System.exit(1);
            }
        });
    }

實現(xiàn)監(jiān)聽類，處理具體邏輯

/**
 * 延遲消息監(jiān)聽
 * 
 */
@Component
public class CourseOrderTimeoutListener implements ApplicationListener<ApplicationReadyEvent>
{
 
    @Resource
    private MQUtil mqUtil;
 
    @Resource
    private CourseOrderTimeoutHandler courseOrderTimeoutHandler;
 
    @Override
    public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent)
{
        // 訂單超時監(jiān)聽
        mqUtil.messageListener(EnumTopic.ORDER_TIMEOUT, EnumGroup.ORDER_TIMEOUT_GROUP, courseOrderTimeoutHandler);
    }
}

/**
 *  實現(xiàn)監(jiān)聽
 */
@Slf4j
@Component
public class CourseOrderTimeoutHandler implements MessageListenerConcurrently
{
 
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for (MessageExt msg : list)
        {
            // 得到消息體
            String body = new String(msg.getBody());
            JSONObject userJson = JSONObject.parseObject(body);
            TCourseBuy courseBuyDetails = JSON.toJavaObject(userJson, TCourseBuy.class);
 
            // 處理具體的業(yè)務(wù)邏輯，，，，，
 
      DateFormat format =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
           log.info("消費時間：{}", format.format(new Date()));
           
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

這種方式相比定時任務(wù)好了很多，但是有一個致命的缺點，就是延遲等級只有18種（商業(yè)版本支持自定義時間），如果我們想把關(guān)閉訂單時間設(shè)置在15分鐘該如何處理呢？顯然不夠靈活。

三、rabbitmq死信隊列的方式

Rabbitmq本身是沒有延遲隊列的，只能通過Rabbitmq本身隊列的特性來實現(xiàn)，想要Rabbitmq實現(xiàn)延遲隊列，需要使用Rabbitmq的死信交換機（Exchange）和消息的存活時間TTL（Time To Live）

死信交換機 一個消息在滿足如下條件下，會進死信交換機，記住這里是交換機而不是隊列，一個交換機可以對應(yīng)很多隊列。

一個消息被Consumer拒收了，并且reject方法的參數(shù)里requeue是false。也就是說不會被再次放在隊列里，被其他消費者使用。上面的消息的TTL到了，消息過期了。

隊列的長度限制滿了。排在前面的消息會被丟棄或者扔到死信路由上。 死信交換機就是普通的交換機，只是因為我們把過期的消息扔進去，所以叫死信交換機，并不是說死信交換機是某種特定的交換機

消息TTL（消息存活時間） 消息的TTL就是消息的存活時間。RabbitMQ可以對隊列和消息分別設(shè)置TTL。對隊列設(shè)置就是隊列沒有消費者連著的保留時間，也可以對每一個單獨的消息做單獨的設(shè)置。超過了這個時間，我們認為這個消息就死了，稱之為死信。如果隊列設(shè)置了，消息也設(shè)置了，那么會取值較小的。所以一個消息如果被路由到不同的隊列中，這個消息死亡的時間有可能不一樣（不同的隊列設(shè)置）。這里單講單個消息的TTL，因為它才是實現(xiàn)延遲任務(wù)的關(guān)鍵。

byte[] messageBodyBytes = "Hello, world!".getBytes();  
AMQP.BasicProperties properties = new AMQP.BasicProperties();  
properties.setExpiration("60000");  
channel.basicPublish("my-exchange", "queue-key", properties, messageBodyBytes);

可以通過設(shè)置消息的expiration字段或者x-message-ttl屬性來設(shè)置時間，兩者是一樣的效果。只是expiration字段是字符串參數(shù)，所以要寫個int類型的字符串：當上面的消息扔到隊列中后，過了60秒，如果沒有被消費，它就死了。不會被消費者消費到。這個消息后面的，沒有“死掉”的消息對頂上來，被消費者消費。死信在隊列中并不會被刪除和釋放，它會被統(tǒng)計到隊列的消息數(shù)中去

處理流程圖

創(chuàng)建交換機（Exchanges）和隊列（Queues）

創(chuàng)建死信交換機

如圖所示，就是創(chuàng)建一個普通的交換機，這里為了方便區(qū)分，把交換機的名字取為：delay

創(chuàng)建自動過期消息隊列 這個隊列的主要作用是讓消息定時過期的，比如我們需要2小時候關(guān)閉訂單，我們就需要把消息放進這個隊列里面，把消息過期時間設(shè)置為2小時

創(chuàng)建一個一個名為delay_queue1的自動過期的隊列，當然圖片上面的參數(shù)并不會讓消息自動過期，因為我們并沒有設(shè)置x-message-ttl參數(shù)，如果整個隊列的消息有消息都是相同的，可以設(shè)置，這里為了靈活，所以并沒有設(shè)置，另外兩個參數(shù)x-dead-letter-exchange代表消息過期后，消息要進入的交換機，這里配置的是delay，也就是死信交換機，x-dead-letter-routing-key是配置消息過期后，進入死信交換機的routing-key,跟發(fā)送消息的routing-key一個道理，根據(jù)這個key將消息放入不同的隊列

創(chuàng)建消息處理隊列 這個隊列才是真正處理消息的隊列，所有進入這個隊列的消息都會被處理

消息隊列的名字為delay_queue2 消息隊列綁定到交換機進入交換機詳情頁面，將創(chuàng)建的2個隊列（delayqueue1和delayqueue2）綁定到交換機上面

自動過期消息隊列的routing key 設(shè)置為delay 綁定delayqueue2

delayqueue2 的key要設(shè)置為創(chuàng)建自動過期的隊列的x-dead-letter-routing-key參數(shù)，這樣當消息過期的時候就可以自動把消息放入delay_queue2這個隊列中了綁定后的管理頁面如下圖：

當然這個綁定也可以使用代碼來實現(xiàn)，只是為了直觀表現(xiàn)，所以本文使用的管理平臺來操作 發(fā)送消息

String msg = "hello word";  
MessageProperties messageProperties = newMessageProperties();  
messageProperties.setExpiration("6000");
messageProperties.setCorrelationId(UUID.randomUUID().toString().getBytes());
Message message = newMessage(msg.getBytes(), messageProperties);
rabbitTemplate.convertAndSend("delay", "delay",message);

設(shè)置了讓消息6秒后過期注意：因為要讓消息自動過期，所以一定不能設(shè)置delay_queue1的監(jiān)聽，不能讓這個隊列里面的消息被接受到，否則消息一旦被消費，就不存在過期了

接收消息 接收消息配置好delay_queue2的監(jiān)聽就好了

package wang.raye.rabbitmq.demo1;
import org.springframework.amqp.core.AcknowledgeMode;  
import org.springframework.amqp.core.Binding;  
import org.springframework.amqp.core.BindingBuilder;  
import org.springframework.amqp.core.DirectExchange;  
import org.springframework.amqp.core.Message;  
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.ChannelAwareMessageListener;  
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;  
import org.springframework.beans.factory.annotation.Autowired;  
import org.springframework.context.annotation.Bean;  
import org.springframework.context.annotation.Configuration;
@Configuration
publicclassDelayQueue{  
    /** 消息交換機的名字*/
    publicstaticfinalString EXCHANGE = "delay";
    /** 隊列key1*/
    publicstaticfinalString ROUTINGKEY1 = "delay";
    /** 隊列key2*/
    publicstaticfinalString ROUTINGKEY2 = "delay_key";
    /**
     * 配置鏈接信息
     * @return
     */
    @Bean
    publicConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = newCachingConnectionFactory("120.76.237.8",5672);
        connectionFactory.setUsername("kberp");
        connectionFactory.setPassword("kberp");
        connectionFactory.setVirtualHost("/");
        connectionFactory.setPublisherConfirms(true); // 必須要設(shè)置
        return connectionFactory;
    }
    /**  
     * 配置消息交換機
     * 針對消費者配置  
        FanoutExchange: 將消息分發(fā)到所有的綁定隊列，無routingkey的概念  
        HeadersExchange ：通過添加屬性key-value匹配  
        DirectExchange:按照routingkey分發(fā)到指定隊列  
        TopicExchange:多關(guān)鍵字匹配  
     */  
    @Bean  
    publicDirectExchange defaultExchange() {  
        returnnewDirectExchange(EXCHANGE, true, false);
    }
    /**
     * 配置消息隊列2
     * 針對消費者配置  
     * @return
     */
    @Bean
    publicQueue queue() {  
       returnnewQueue("delay_queue2", true); //隊列持久  
    }
    /**
     * 將消息隊列2與交換機綁定
     * 針對消費者配置  
     * @return
     */
    @Bean  
    @Autowired
    publicBinding binding() {  
        returnBindingBuilder.bind(queue()).to(defaultExchange()).with(DelayQueue.ROUTINGKEY2);  
    }
    /**
     * 接受消息的監(jiān)聽，這個監(jiān)聽會接受消息隊列1的消息
     * 針對消費者配置  
     * @return
     */
    @Bean  
    @Autowired
    publicSimpleMessageListenerContainer messageContainer2(ConnectionFactory connectionFactory) {  
        SimpleMessageListenerContainer container = newSimpleMessageListenerContainer(connectionFactory());  
        container.setQueues(queue());  
        container.setExposeListenerChannel(true);  
        container.setMaxConcurrentConsumers(1);  
        container.setConcurrentConsumers(1);  
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //設(shè)置確認模式手工確認  
        container.setMessageListener(newChannelAwareMessageListener() {
            publicvoid onMessage(Message message, com.rabbitmq.client.Channel channel) throwsException{
                byte[] body = message.getBody();  
                System.out.println("delay_queue2 收到消息 : "+ newString(body));  
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //確認消息成功消費  
            }  
        });  
        return container;  
    }  
}

這種方式可以自定義進入死信隊列的時間；是不是很完美，但是有的小伙伴的情況是消息中間件就是rocketmq，公司也不可能會用商業(yè)版，怎么辦？那就進入下一節(jié)

四、時間輪算法

(1)創(chuàng)建環(huán)形隊列，例如可以創(chuàng)建一個包含3600個slot的環(huán)形隊列(本質(zhì)是個數(shù)組)

(2)任務(wù)集合，環(huán)上每一個slot是一個Set 同時，啟動一個timer，這個timer每隔1s，在上述環(huán)形隊列中移動一格，有一個Current Index指針來標識正在檢測的slot。

Task結(jié)構(gòu)中有兩個很重要的屬性： (1)Cycle-Num：當Current Index第幾圈掃描到這個Slot時，執(zhí)行任務(wù) (2)訂單號，要關(guān)閉的訂單號（也可以是其他信息，比如：是一個基于某個訂單號的任務(wù)）

假設(shè)當前Current Index指向第0格，例如在3610秒之后，有一個訂單需要關(guān)閉，只需： (1)計算這個訂單應(yīng)該放在哪一個slot，當我們計算的時候現(xiàn)在指向1，3610秒之后，應(yīng)該是第10格，所以這個Task應(yīng)該放在第10個slot的Set中 (2)計算這個Task的Cycle-Num，由于環(huán)形隊列是3600格(每秒移動一格，正好1小時)，這個任務(wù)是3610秒后執(zhí)行，所以應(yīng)該繞3610/3600=1圈之后再執(zhí)行，于是Cycle-Num=1

Current Index不停的移動，每秒移動到一個新slot，這個slot中對應(yīng)的Set，每個Task看Cycle-Num是不是0： (1)如果不是0，說明還需要多移動幾圈，將Cycle-Num減1 (2)如果是0，說明馬上要執(zhí)行這個關(guān)單Task了，取出訂單號執(zhí)行關(guān)單(可以用單獨的線程來執(zhí)行Task)，并把這個訂單信息從Set中刪除即可。 (1)無需再輪詢?nèi)坑唵?，效率?(2)一個訂單，任務(wù)只執(zhí)行一次 (3)時效性好，精確到秒(控制timer移動頻率可以控制精度)

五、redis過期監(jiān)聽

1.修改redis.windows.conf配置文件中notify-keyspace-events的值 默認配置notify-keyspace-events的值為 "" 修改為 notify-keyspace-events Ex 這樣便開啟了過期事件

2. 創(chuàng)建配置類RedisListenerConfig（配置RedisMessageListenerContainer這個Bean）

package com.zjt.shop.config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
@Configuration
public class RedisListenerConfig {
    @Autowired
    private RedisTemplate redisTemplate;
    /**
     * @return
     */
    @Bean
    public RedisTemplate redisTemplateInit() {
        // key序列化
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        //val實例化
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        return redisTemplate;
    }
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        return container;
    }
}

3.繼承KeyExpirationEventMessageListener創(chuàng)建redis過期事件的監(jiān)聽類

package com.zjt.shop.common.util;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.zjt.shop.modules.order.service.OrderInfoService;
import com.zjt.shop.modules.product.entity.OrderInfoEntity;
import com.zjt.shop.modules.product.mapper.OrderInfoMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
    public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
        super(listenerContainer);
    }
    @Autowired
    private OrderInfoMapper orderInfoMapper;
    /**
     * 針對redis數(shù)據(jù)失效事件，進行數(shù)據(jù)處理
     * @param message
     * @param pattern
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
      try {
          String key = message.toString();
          //從失效key中篩選代表訂單失效的key
          if (key != null && key.startsWith("order_")) {
              //截取訂單號，查詢訂單，如果是未支付狀態(tài)則為-取消訂單
              String orderNo = key.substring(6);
              QueryWrapper<OrderInfoEntity> queryWrapper = new QueryWrapper<>();
              queryWrapper.eq("order_no",orderNo);
              OrderInfoEntity orderInfo = orderInfoMapper.selectOne(queryWrapper);
              if (orderInfo != null) {
                  if (orderInfo.getOrderState() == 0) {   //待支付
                      orderInfo.setOrderState(4);         //已取消
                      orderInfoMapper.updateById(orderInfo);
                      log.info("訂單號為【" + orderNo + "】超時未支付-自動修改為已取消狀態(tài)");
                  }
              }
          }
      } catch (Exception e) {
          e.printStackTrace();
          log.error("【修改支付訂單過期狀態(tài)異?！浚? + e.getMessage());
      }
    }
}

4：測試 通過redis客戶端存一個有效時間為3s的訂單：

結(jié)果：

到此這篇關(guān)于Android中關(guān)于定時任務(wù)實現(xiàn)關(guān)閉訂單問題的文章就介紹到這了,更多相關(guān)android 定時任務(wù)關(guān)閉訂單內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: