《通過(guò)DelayQueue實(shí)現(xiàn)延時(shí)任務(wù)》

《基于netty時(shí)間輪算法實(shí)戰(zhàn)》

這兩種方法都有一個(gè)缺點(diǎn)：都是基于單體應(yīng)用的內(nèi)存的方式運(yùn)行延時(shí)任務(wù)的，一旦出現(xiàn)單點(diǎn)故障，可能出現(xiàn)延時(shí)任務(wù)數(shù)據(jù)的丟失。所以此篇文章給大家介紹實(shí)現(xiàn)延時(shí)任務(wù)的第三種方式，結(jié)合redis zset實(shí)現(xiàn)延時(shí)任務(wù)，可以解決單點(diǎn)故障的問(wèn)題。給出實(shí)現(xiàn)原理、完整實(shí)現(xiàn)代碼，以及這種實(shí)現(xiàn)方式的優(yōu)缺點(diǎn)。

一、實(shí)現(xiàn)原理

首先來(lái)介紹一下實(shí)現(xiàn)原理，我們需要使用redis zset來(lái)實(shí)現(xiàn)延時(shí)任務(wù)的需求，所以我們需要知道zset的應(yīng)用特性。zset作為redis的有序集合數(shù)據(jù)結(jié)構(gòu)存在，排序的依據(jù)就是score。

所以我們可以利用zset score這個(gè)排序的這個(gè)特性，來(lái)實(shí)現(xiàn)延時(shí)任務(wù)

• 在用戶(hù)下單的時(shí)候，同時(shí)生成延時(shí)任務(wù)放入redis，key是可以自定義的，比如：delaytask:order
• value的值分成兩個(gè)部分，一個(gè)部分是score用于排序，一個(gè)部分是member，member的值我們?cè)O(shè)置為訂單對(duì)象（如：訂單編號(hào)），因?yàn)楹罄m(xù)延時(shí)任務(wù)時(shí)效達(dá)成的時(shí)候，我們需要有一些必要的訂單信息（如：訂單編號(hào)），才能完成訂單自動(dòng)取消關(guān)閉的動(dòng)作。
• 「延時(shí)任務(wù)實(shí)現(xiàn)的重點(diǎn)來(lái)了，score我們?cè)O(shè)置為：訂單生成時(shí)間 + 延時(shí)時(shí)長(zhǎng)」。這樣redis會(huì)對(duì)zset按照score延時(shí)時(shí)間進(jìn)行排序。
• 開(kāi)啟redis掃描任務(wù)，獲取"當(dāng)前時(shí)間 > score"的延時(shí)任務(wù)并執(zhí)行。即：當(dāng)前時(shí)間 >  訂單生成時(shí)間 + 延時(shí)時(shí)長(zhǎng)的時(shí)候 ，執(zhí)行延時(shí)任務(wù)。

二、準(zhǔn)備工作

使用 redis zset 這個(gè)方案來(lái)完成延時(shí)任務(wù)的需求，首先肯定是需要redis，這一點(diǎn)毫無(wú)疑問(wèn)。redis的搭建網(wǎng)上有很多的文章，我這里就不贅述了。

其次，筆者長(zhǎng)期的java類(lèi)應(yīng)用系統(tǒng)開(kāi)發(fā)都是使用SpringBoot來(lái)完成，所以也是習(xí)慣使用SpringBoot的redis集成方案。首先通過(guò)maven坐標(biāo)引入spring-boot-starter-data-redis

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
</dependency>

其次需要在Spring Boot的application.yml配置文件中，配置redis數(shù)據(jù)庫(kù)的鏈接信息。我這里配置的是redis的單例，如果大家的生產(chǎn)環(huán)境是哨兵模式、或者是集群模式的redis，這里的配置方式需要進(jìn)行微調(diào)。其實(shí)這部分內(nèi)容在我的個(gè)人博客里面都曾經(jīng)系統(tǒng)的介紹過(guò)，感興趣的朋友可以關(guān)注我的個(gè)人博客。

spring:
  redis:
    database: 0 # Redis 數(shù)據(jù)庫(kù)索引（默認(rèn)為 0）
    host: 192.168.161.3 # Redis 服務(wù)器地址
    port: 6379 # Redis 服務(wù)器連接端口
    password: 123456 # Redis 服務(wù)器連接密碼（默認(rèn)為空）
    timeout:  5000  # 連接超時(shí)，單位ms
    lettuce:
      pool:
        max-active: 8 # 連接池最大連接數(shù)（使用負(fù)值表示沒(méi)有限制） 默認(rèn) 8
        max-wait: -1 # 連接池最大阻塞等待時(shí)間（使用負(fù)值表示沒(méi)有限制） 默認(rèn) -1
        max-idle: 8 # 連接池中的最大空閑連接 默認(rèn) 8
        min-idle: 0 # 連接池中的最小空閑連接 默認(rèn) 0

三、代碼實(shí)現(xiàn)

下面的這個(gè)類(lèi)就是延時(shí)任務(wù)的核心實(shí)現(xiàn)了，一共包含三個(gè)核心方法，我們來(lái)一一說(shuō)明一下：

• produce方法，用于生成訂單-order為訂單信息，可以是訂單流水號(hào)，用于延時(shí)任務(wù)達(dá)到時(shí)效后關(guān)閉訂單
• afterPropertiesSet方法是InitializingBean接口的方法，之所以實(shí)現(xiàn)這個(gè)接口，是因?yàn)槲覀冃枰趹?yīng)用啟動(dòng)的時(shí)候開(kāi)啟redis掃描任務(wù)。即：當(dāng)OrderDelayService bean初始化的時(shí)候，開(kāi)啟redis掃描任務(wù)循環(huán)獲取延時(shí)任務(wù)數(shù)據(jù)。
• consuming函數(shù)，用于從redis獲取延時(shí)任務(wù)數(shù)據(jù)，消費(fèi)延時(shí)任務(wù)，執(zhí)行超時(shí)訂單關(guān)閉等操作。為了避免阻塞for循環(huán)，影響后面延時(shí)任務(wù)的執(zhí)行，所以這個(gè)consuming函數(shù)一定要做成異步的，參考Spring Boot異步任務(wù)及Async注解的使用方法。我之前寫(xiě)過(guò)一個(gè)SpringBoot的**「可觀測(cè)、易配置」**的異步任務(wù)線(xiàn)程池開(kāi)源項(xiàng)目，源代碼地址：https://gitee.com/hanxt/zimug-monitor-threadpool  。我的這個(gè)zimug-monitor-threadpool開(kāi)源項(xiàng)目，可以做到對(duì)線(xiàn)程池使用情況的監(jiān)控，我自己平時(shí)用的效果還不錯(cuò)，向大家推薦一下！

@Component
public class OrderDelayService  implements InitializingBean {
  //redis zset key
  public static final String ORDER_DELAY_TASK_KEY = "delaytask:order";

  @Resource
  private StringRedisTemplate stringRedisTemplate;

  //生成訂單-order為訂單信息，可以是訂單流水號(hào)，用于延時(shí)任務(wù)達(dá)到時(shí)效后關(guān)閉訂單
  public void produce(String orderSerialNo){
    stringRedisTemplate.opsForZSet().add(
            ORDER_DELAY_TASK_KEY,     // redis key
            orderSerialNo,    // zset  member
            //30分鐘延時(shí)
            System.currentTimeMillis() + (30 * 60 * 1000)    //zset score
    );
  }

  //延時(shí)任務(wù)，也是異步任務(wù)，延時(shí)任務(wù)達(dá)到時(shí)效之后關(guān)閉訂單，并將延時(shí)任務(wù)從redis zset刪除
  @Async("test")
  public void consuming(){
       
      Set<ZSetOperations.TypedTuple<String>> orderSerialNos = stringRedisTemplate.opsForZSet().rangeByScoreWithScores(
              ORDER_DELAY_TASK_KEY,
              0,  //延時(shí)任務(wù)score最小值
              System.currentTimeMillis() //延時(shí)任務(wù)score最大值（當(dāng)前時(shí)間）
      );
      if (!CollectionUtils.isEmpty(orderSerialNos)) {
        for (ZSetOperations.TypedTuple<String> orderSerialNo : orderSerialNos) {
          //這里根據(jù)orderSerialNo去檢查用戶(hù)是否完成了訂單支付
          //如果用戶(hù)沒(méi)有支付訂單，去執(zhí)行訂單關(guān)閉的操作
          System.out.println("訂單" + orderSerialNo.getValue() + "超時(shí)被自動(dòng)關(guān)閉");
          //訂單關(guān)閉之后，將訂單延時(shí)任務(wù)從隊(duì)列中刪除
          stringRedisTemplate.opsForZSet().remove(ORDER_DELAY_TASK_KEY, orderSerialNo.getValue());
        }
      }
  }

  //該類(lèi)對(duì)象Bean實(shí)例化之后，就開(kāi)啟while掃描任務(wù)
  @Override
  public void afterPropertiesSet() throws Exception {
    new Thread(() -> {  //開(kāi)啟新的線(xiàn)程，否則SpringBoot應(yīng)用初始化無(wú)法啟動(dòng)
      while(true){
        try {
          Thread.sleep(5 * 1000);   //每5秒掃描一次redis庫(kù)獲取延時(shí)數(shù)據(jù)，不用太頻繁沒(méi)必要
        } catch (InterruptedException e) {
          e.printStackTrace();  //本文只是示例，生產(chǎn)環(huán)境請(qǐng)做好相關(guān)的異常處理
        }
        consuming();
      }
    }).start();
  }
}

更多的內(nèi)容參考代碼中的注釋?zhuān)枰P(guān)注的點(diǎn)是：

• 上文中的rangeByScoreWithScores方法用于從redis中獲取延時(shí)任務(wù)，score大于0小于當(dāng)前時(shí)間的所有延時(shí)任務(wù)，都將被從redis里面取出來(lái)。每5秒執(zhí)行一次，所以延時(shí)任務(wù)的誤差不會(huì)超過(guò)5秒。
• 上文中的訂單信息，我只保留了訂單唯一流水號(hào)，用于關(guān)閉訂單。如果你的業(yè)務(wù)需要傳遞更多的訂單信息，請(qǐng)使用RedisTemplate操作訂單類(lèi)對(duì)象，而不是StringRedisTemplate操作訂單流水號(hào)字符串。
• 訂單下單的時(shí)候，使用如下的方法，將訂單序列號(hào)放入redis zset中即可實(shí)現(xiàn)延時(shí)任務(wù)

orderDelayService.produce("這里填寫(xiě)訂單編號(hào)");

四、優(yōu)缺點(diǎn)

使用redis zset來(lái)實(shí)現(xiàn)延時(shí)任務(wù)的優(yōu)點(diǎn)是：相對(duì)于本文開(kāi)頭介紹的兩種方法，我們的延時(shí)任務(wù)是保存在redis里面的，redis具有數(shù)據(jù)持久化的機(jī)制，可以有效的避免延時(shí)任務(wù)數(shù)據(jù)的丟失。另外，redis還可以通過(guò)哨兵模式、集群模式有效的避免單點(diǎn)故障造成的服務(wù)中斷。至于缺點(diǎn)嘛，我覺(jué)得沒(méi)什么缺點(diǎn)。如果非要勉強(qiáng)的說(shuō)一個(gè)缺點(diǎn)的話(huà)，那就是我們需要額外維護(hù)redis服務(wù)，增加了硬件資源的需求和運(yùn)維成本。但是現(xiàn)在隨著微服務(wù)的興起，redis幾乎已經(jīng)成了應(yīng)用系統(tǒng)的標(biāo)配，redis復(fù)用即可，所以我感覺(jué)這也算不上什么缺點(diǎn)吧！

到此這篇關(guān)于Java利用redis zset實(shí)現(xiàn)延時(shí)任務(wù)詳解的文章就介紹到這了,更多相關(guān)redis zset延時(shí)任務(wù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！