1. <dd id="erndk"></dd>
                1. 說下Redis 延時任務的方案

                  互聯網 2022/5/2 19:13:16

                  引言 在開發中,往往會遇到一些關于延時任務的需求。例如生成訂單30分鐘未支付,則自動取消 生成訂單60秒后,給用戶發短信對上述的任務,我們給一個專業的名字來形容,那就是延時任務。那么這里就會產生一個問題,這個延時任務和定時任務的區別究竟在哪里呢?一共有如下幾…

                  引言

                  在開發中,往往會遇到一些關于延時任務的需求。例如

                  • 生成訂單30分鐘未支付,則自動取消
                  • 生成訂單60秒后,給用戶發短信

                  對上述的任務,我們給一個專業的名字來形容,那就是延時任務。那么這里就會產生一個問題,這個延時任務和定時任務的區別究竟在哪里呢?一共有如下幾點區別

                  • 定時任務有明確的觸發時間,延時任務沒有
                  • 定時任務有執行周期,而延時任務在某事件觸發后一段時間內執行,沒有執行周期
                  • 定時任務一般執行的是批處理操作是多個任務,而延時任務一般是單個任務

                  下面,我們以判斷訂單是否超時為例,進行方案分析

                  方案分析

                  (1)數據庫輪詢

                  思路

                  該方案通常是在小型項目中使用,即通過一個線程定時的去掃描數據庫,通過訂單時間來判斷是否有超時的訂單,然后進行update或delete等操作

                  實現

                  博主當年早期是用quartz來實現的(實習那會的事),簡單介紹一下

                  maven項目引入一個依賴如下所示

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

                  調用Demo類MyJob如下所示

                  package com.rjzheng.delay1;

                  import org.quartz.Job;
                  import org.quartz.JobBuilder;
                  import org.quartz.JobDetail;
                  import org.quartz.JobExecutionContext;
                  import org.quartz.JobExecutionException;
                  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;


                  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)JDK的延遲隊列

                  思路

                  該方案是利用JDK自帶的DelayQueue來實現,這是一個無界阻塞隊列,該隊列只有在延遲期滿的時候才能從中獲取元素,放入DelayQueue中的對象,是必須實現Delayed接口的。

                  DelayedQueue實現工作流程如下圖所示

                  其中

                  • Poll():獲取并移除隊列的超時元素,沒有則返回空
                  • take():獲取并移除隊列的超時元素,如果沒有則wait當前線程,直到有元素滿足超時條件,返回結果。

                  實現

                  定義一個類OrderDelay實現Delayed,代碼如下

                  package com.rjzheng.delay2;

                  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 + "編號的訂單要刪除啦。。。。");
                      }
                  }

                  運行的測試Demo為,我們設定延遲時間為3秒

                  package com.rjzheng.delay2;
                  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 = newDelayQueue < 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();
                              }
                          }
                      }
                  }

                  輸出如下

                  00000001編號的訂單要刪除啦。。。。

                  After 3003 MilliSeconds

                  00000002編號的訂單要刪除啦。。。。

                  After 6006 MilliSeconds

                  00000003編號的訂單要刪除啦。。。。

                  After 9006 MilliSeconds

                  00000004編號的訂單要刪除啦。。。。

                  After 12008 MilliSeconds

                  00000005編號的訂單要刪除啦。。。。

                  After 15009 MilliSeconds

                  可以看到都是延遲3秒,訂單被刪除

                  優缺點

                  優點:

                  • 效率高,任務觸發時間延遲低。

                  缺點:

                  • 服務器重啟后,數據全部消失,怕宕機
                  • 集群擴展相當麻煩
                  • 因為內存條件限制的原因,比如下單未付款的訂單數太多,那么很容易就出現OOM異常
                  • 代碼復雜度較高

                  (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來實現

                  給Pom加上下面的依賴

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

                  測試代碼HashedWheelTimerTest如下所示

                  package com.rjzheng.delay3;
                  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異常

                  (4)redis緩存

                  思路

                  利用redis的zset,zset是一個有序集合,每一個元素(member)都關聯了一個score,通過score排序來取集合中的值

                  • 添加元素:ZADD key score member [[score member] [score member] …]
                  • 按順序查詢元素:ZRANGE key start stop [WITHSCORES]
                  • 查詢元素:score:ZSCORE key member
                  • 移除元素:ZREM key member [member …]

                  測試如下

                  # 添加單個元素
                  redis> ZADD page_rank 10 google.com

                  (integer) 1


                  # 添加多個元素
                  redis> ZADD page_rank 9 baidu.com 8 bing.com

                  (integer) 2

                  redis> ZRANGE page_rank 0 -1 WITHSCORES

                  1) "bing.com"
                  2) "8"
                  3) "baidu.com"
                  4) "9"
                  5) "google.com"
                  6) "10"

                  # 查詢元素的score值

                  redis> ZSCORE page_rank bing.com
                  "8"

                  # 移除單個元素
                   
                  redis> ZREM page_rank google.com

                  (integer) 1

                  redis> ZRANGE page_rank 0 -1 WITHSCORES

                  1) "bing.com"
                  2) "8"
                  3) "baidu.com"
                  4) "9"

                  那么如何實現呢?我們將訂單超時時間戳與訂單號分別設置為score和member,系統掃描第一個元素判斷是否超時,具體如下圖所示

                  實現一

                  package com.rjzheng.delay4;
                  import java.util.Calendar;
                  import java.util.Set;
                  import redis.clients.jedis.Jedis;
                  import redis.clients.jedis.JedisPool;
                  import redis.clients.jedis.Tuple;
                  public class AppTest
                  {
                   private static final String ADDR = "127.0.0.1";
                   private static final int PORT = 6379;
                   private static JedisPool jedisPool = new JedisPool(ADDR, PORT);
                   public static Jedis getJedis()
                    {
                     return jedisPool.getResource();
                    }
                    //生產者,生成5個訂單放進去
                   public void productionDelayMessage()
                    {
                     for(int i = 0; i < 5; i++)
                     {
                      //延遲3秒
                      Calendar cal1 = Calendar.getInstance();
                      cal1.add(Calendar.SECOND, 3);
                      int second3later = (int)(cal1.getTimeInMillis() / 1000);
                      AppTest.getJedis().zadd("OrderId", second3later, "OID0000001" + i);
                      System.out.println(System.currentTimeMillis() + "ms:redis生成了一個訂單任務:訂單ID為" + "OID0000001" + i);
                     }
                    }
                    //消費者,取訂單
                   public void consumerDelayMessage()
                   {
                    Jedis jedis = AppTest.getJedis();
                    while(true)
                    {
                     Set < Tuple > items = jedis.zrangeWithScores("OrderId", 0, 1);
                     if(items == null || items.isEmpty())
                     {
                      System.out.println("當前沒有等待的任務");
                      try
                      {
                       Thread.sleep(500);
                      }
                      catch(InterruptedException e)
                      {
                       // TODO Auto-generated catch block
                       e.printStackTrace();
                      }
                      continue;
                     }
                     int score = (int)((Tuple) items.toArray()[0]).getScore();
                     Calendar cal = Calendar.getInstance();
                     int nowSecond = (int)(cal.getTimeInMillis() / 1000);
                     if(nowSecond >= score)
                     {
                      String orderId = ((Tuple) items.toArray()[0]).getElement();
                      jedis.zrem("OrderId", orderId);
                      System.out.println(System.currentTimeMillis() + "ms:redis消費了一個任務:消費的訂單OrderId為" + orderId);
                     }
                    }
                   }
                   public static void main(String[] args)
                   {
                    AppTest appTest = new AppTest();
                    appTest.productionDelayMessage();
                    appTest.consumerDelayMessage();
                   }
                  }

                  此時對應輸出如下

                  可以看到,幾乎都是3秒之后,消費訂單。

                  然而,這一版存在一個致命的硬傷,在高并發條件下,多消費者會取到同一個訂單號,我們上測試代碼ThreadTest

                  package com.rjzheng.delay4;
                  import java.util.concurrent.CountDownLatch;
                  public class ThreadTest
                  {
                   private static final int threadNum = 10;
                   private static CountDownLatch cdl = newCountDownLatch(threadNum);
                   static class DelayMessage implements Runnable
                   {
                    public void run()
                    {
                     try
                     {
                      cdl.await();
                     }
                     catch(InterruptedException e)
                     {
                      // TODO Auto-generated catch block
                      e.printStackTrace();
                     }
                     AppTest appTest = new AppTest();
                     appTest.consumerDelayMessage();
                    }
                   }
                   public static void main(String[] args)
                   {
                    AppTest appTest = new AppTest();
                    appTest.productionDelayMessage();
                    for(int i = 0; i < threadNum; i++)
                    {
                     new Thread(new DelayMessage()).start();
                     cdl.countDown();
                    }
                   }
                  }

                  輸出如下所示

                  顯然,出現了多個線程消費同一個資源的情況。

                  解決方案

                  (1)用分布式鎖,但是用分布式鎖,性能下降了,該方案不細說。

                  (2)對ZREM的返回值進行判斷,只有大于0的時候,才消費數據,于是將consumerDelayMessage()方法里的

                  if(nowSecond >= score)
                  {
                   String orderId = ((Tuple) items.toArray()[0]).getElement();
                   jedis.zrem("OrderId", orderId);
                   System.out.println(System.currentTimeMillis() + "ms:redis消費了一個任務:消費的訂單OrderId為" + orderId);
                  }

                  修改為

                  if(nowSecond >= score)
                  {
                   String orderId = ((Tuple) items.toArray()[0]).getElement();
                   Long num = jedis.zrem("OrderId", orderId);
                   if(num != null && num > 0)
                   {
                    System.out.println(System.currentTimeMillis() + "ms:redis消費了一個任務:消費的訂單OrderId為" + orderId);
                   }
                  }

                  在這種修改后,重新運行ThreadTest類,發現輸出正常了

                  思路二

                  該方案使用redis的Keyspace Notifications,中文翻譯就是鍵空間機制,就是利用該機制可以在key失效之后,提供一個回調,實際上是redis會給客戶端發送一個消息。是需要redis版本2.8以上。

                  實現二

                  在redis.conf中,加入一條配置

                  notify-keyspace-events Ex

                  運行代碼如下

                  package com.rjzheng.delay5;
                  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
                   { < ahref = 'http://www.jobbole.com/members/wx610506454' > @Override < /a>
                    public void onMessage(String channel, String message)
                    {
                     System.out.println(System.currentTimeMillis() + "ms:" + message + "訂單取消");
                    }
                   }
                  }

                  輸出如下

                  可以明顯看到3秒過后,訂單取消了

                  ps:redis的pub/sub機制存在一個硬傷,官網內容如下

                  原:

                  Because Redis Pub/Sub is fire and forget currently there is no way to use this feature if your application demands reliable notification of events, that is, if your Pub/Sub client disconnects, and reconnects later, all the events delivered during the time the client was disconnected are lost.

                  翻:

                  Redis的發布/訂閱目前是即發即棄(fire and forget)模式的,因此無法實現事件的可靠通知。也就是說,如果發布/訂閱的客戶端斷鏈之后又重連,則在客戶端斷鏈期間的所有事件都丟失了。

                  因此,方案二不是太推薦。當然,如果你對可靠性要求不高,可以使用。

                  優缺點

                  優點:

                  • 由于使用Redis作為消息通道,消息都存儲在Redis中。如果發送程序或者任務處理程序掛了,重啟之后,還有重新處理數據的可能性。
                  • 做集群擴展相當方便
                  • 時間準確度高

                  缺點:

                  • 需要額外進行redis維護

                  (5)使用消息隊列

                  我們可以采用rabbitMQ的延時隊列。RabbitMQ具有以下兩個特性,可以實現延遲隊列

                  RabbitMQ可以針對Queue和Message設置 x-message-tt,來控制消息的生存時間,如果超時,則消息變為dead letter

                  lRabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key(可選)兩個參數,用來控制隊列內出現了deadletter,則按照這兩個參數重新路由。

                  結合以上兩個特性,就可以模擬出延遲消息的功能,具體的,我改天再寫一篇文章,這里再講下去,篇幅太長。

                  優缺點

                  優點:

                  • 高效,可以利用rabbitmq的分布式特性輕易的進行橫向擴展,消息支持持久化增加了可靠性。

                  缺點:

                  • 本身的易用度要依賴于rabbitMq的運維.因為要引用rabbitMq,所以復雜度和成本變高
                  隨時隨地學軟件編程-關注百度小程序和微信小程序
                  關于找一找教程網

                  本站文章僅代表作者觀點,不代表本站立場,所有文章非營利性免費分享。
                  本站提供了軟件編程、網站開發技術、服務器運維、人工智能等等IT技術文章,希望廣大程序員努力學習,讓我們用科技改變世界。
                  [說下Redis 延時任務的方案]http://www.yachtsalesaustralia.com/tech/detail-318764.html

                  贊(0)
                  關注微信小程序
                  程序員編程王-隨時隨地學編程

                  掃描二維碼或查找【程序員編程王】

                  可以隨時隨地學編程啦!

                  技術文章導航 更多>
                  国产在线拍揄自揄视频菠萝

                        1. <dd id="erndk"></dd>