優雅實現延時任務之Redis篇
什麼是延時任務
延時任務，顧名思義，就是延遲一段時間後才執行的任務。舉個例子，假設我們有個釋出資訊的功能，運營需要在每天早上7點準時釋出資訊，但是早上7點大家都還沒上班，這個時候就可以使用延時任務來實現資訊的延時釋出了。只要在前一天下班前指定第二天要傳送資訊的時間，到了第二天指定的時間點資訊就能準時發出去了。如果大家有運營過公眾號，就會知道公眾號後臺也有文章定時傳送的功能。總而言之，延時任務的使用還是很廣泛的。關於延時任務的實現方式，我知道的就不下於3種，後面會逐一介紹，今天就講下如何用redis實現延時任務。
延時任務的特點
在介紹具體方案之前，我們不妨先想一下要實現一個延時系統，有哪些內容是必須儲存下來的（這裡的儲存不一定是指持久化，也可以是放在記憶體中，取決於延時任務的重要程度）。首先要儲存的就是任務的描述。假如你要處理的延時任務是延時釋出資訊，那麼你至少要儲存資訊的id吧。另外，如果你有多種任務型別，比如：延時推送訊息、延時清洗資料等等，那麼你還需要儲存任務的型別。以上總總，都歸屬於任務描述。除此之外，你還必須儲存任務執行的時間點吧，一般來說就是時間戳。此外，我們還需要根據任務的執行時間進行排序，因為延時任務佇列裡的任務可能會有很多，只有到了時間點的任務才應該被執行，所以必須支援對任務執行時間進行排序。
使用Redis實現延時任務
以上就是一個延遲任務系統必須具備的要素了。回到redis，有什麼資料結構可以既儲存任務描述，又能儲存任務執行時間，還能根據任務執行時間進行排序呢？想來想去，似乎只有Sorted Set了。我們可以把任務的描述序列化成字串，放在Sorted Set的value中，然後把任務的執行時間戳作為score，利用Sorted Set天然的排序特性，執行時刻越早的會排在越前面。這樣一來，我們只要開一個或多個定時執行緒，每隔一段時間去查一下這個Sorted Set中score小於或等於當前時間戳的元素（這可以通過zrangebyscore命令實現），然後再執行元素對應的任務即可。當然，執行完任務後，還要將元素從Sorted Set中刪除，避免任務重複執行。如果是多個執行緒去輪詢這個Sorted Set，還有考慮併發問題，假如說一個任務到期了，也被多個執行緒拿到了，這個時候必須保證只有一個執行緒能執行這個任務，這可以通過zrem命令來實現，只有刪除成功了，才能執行任務，這樣就能保證任務不被多個任務重複執行了。
接下來看程式碼。首先看下專案結構：
一共4個類：Constants類定義了Redis key相關的常量。DelayTaskConsumer是延時任務的消費者，這個類負責從Redis拉取到期的任務，並封裝了任務消費的邏輯。DelayTaskProducer則是延時任務的生產者，主要用於將延時任務放到Redis中。RedisClient則是Redis客戶端的工具類。
最主要的類就是DelayTaskConsumer和DelayTaskProducer了。
我們先來看下生產者DelayTaskProducer：
public class DelayTaskProducer {
public void produce(String newsId,long timeStamp){
Jedis client = RedisClient.getClient();
try {
client.zadd(Constants.DELAY_TASK_QUEUE,timeStamp,newsId);
}finally {
client.close();
}
}
}
複製程式碼
程式碼很簡單，就是將任務描述（為了方便，這裡只儲存資訊的id）和任務執行的時間戳放到Redis的Sorted Set中。
接下來是延時任務的消費者DelayTaskConsumer：
public class DelayTaskConsumer {
private ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
public void start(){
scheduledExecutorService.scheduleWithFixedDelay(new DelayTaskHandler(),1,1, TimeUnit.SECONDS);
}
public static class DelayTaskHandler implements Runnable{
@Override
public void run() {
Jedis client = RedisClient.getClient();
try {
Set<String> ids = client.zrangeByScore(Constants.DELAY_TASK_QUEUE, 0, System.currentTimeMillis(),
0, 1);
if(ids==null||ids.isEmpty()){
return;
}
for(String id:ids){
Long count = client.zrem(Constants.DELAY_TASK_QUEUE, id);
if(count!=null&&count==1){
System.out.println(MessageFormat.format("釋出資訊。id - {0} , timeStamp - {1} , " +
"threadName - {2}",id,System.currentTimeMillis(),Thread.currentThread().getName()));
}
}
}finally {
client.close();
}
}
}
}
複製程式碼
首先看start方法。在這個方法裡面我們利用Java的ScheduledExecutorService開了一個排程執行緒池，這個執行緒池會每隔1秒鐘排程DelayTaskHandler中的run方法。
DelayTaskHandler類就是具體的排程邏輯了。主要有2個步驟，一個是從Redis Sorted Set中拉取到期的延時任務，另一個是執行到期的延時任務。拉取到期的延時任務是通過zrangeByScore命令實現的，處理多執行緒併發問題是通過zrem命令實現的。程式碼不復雜，這裡就不多做解釋了。
接下來測試一下：
public class DelayTaskTest {
public static void main(String[] args) {
DelayTaskProducer producer=new DelayTaskProducer();
long now=new Date().getTime();
System.out.println(MessageFormat.format("start time - {0}",now));
producer.produce("1",now+ TimeUnit.SECONDS.toMillis(5));
producer.produce("2",now+TimeUnit.SECONDS.toMillis(10));
producer.produce("3",now+ TimeUnit.SECONDS.toMillis(15));
producer.produce("4",now+TimeUnit.SECONDS.toMillis(20));
for(int i=0;i<10;i++){
new DelayTaskConsumer().start();
}
}
}
複製程式碼
我們首先生產了4個延時任務，執行時間分別是程式開始執行後的5秒、10秒、15秒、20秒，然後啟動了10個消費者去消費延時任務。執行效果如下：
可以看到，任務確實能夠在相應的時間點左右被執行，不過有少許時間誤差，這個是因為我們拉取到期任務是通過定時任務拉取而不是實時推送的，而且拉取任務時有一部分網路開銷，再者，我們的任務處理邏輯是同步處理的，需要上一次的任務處理完，才能拉取下一批任務，這些因素都會造成延時任務的執行時間產生偏差。
總結
以上就是通過Redis實現延時任務的思路了。這裡提供的只是一個最簡單的版本，實際上還有很多地方可以優化。比如，我們可以把任務的處理邏輯再放到單獨的執行緒池中去執行，這樣的話任務消費者只需要負責任務的排程就可以了，好處就是可以減少任務執行時間偏差。還有就是，這裡為了方便，任務的描述儲存的只是任務的id，如果有多種不同型別的任務，像前面說的傳送資訊任務和推送訊息任務，那麼就要通過額外儲存任務的型別來進行區分，或者使用不同的Sorted Set來存放延時任務了。
除此之外，上面的例子每次拉取延時任務時，只拉取1個，如果說某一個時刻要處理的任務數非常多，那麼會有一部分任務延遲比較嚴重，這裡可以優化成每次拉取不止1個到期的任務，比如說10個，然後再逐個進行處理，這樣的話可以極大地提升排程效率，因為如果是使用上面的方法，拉取10個任務需要10次排程，每次間隔1秒，總共需要10秒才能把10個任務拉取完，如果改成一次拉取10個，只需要1次就能完成了，效率提升還是挺大的。
當然還可以從另一個角度來優化。大家看上面的程式碼，當拉取到待執行任務時，就直接執行任務，任務執行完該執行緒也就退出了，但是這個時候，佇列裡可能還有很多待執行的任務（因為我們拉取任務時，限制了拉取的數量），所以其實在這裡可以使用迴圈，當拉取不到待執行任務時，才結束排程，當有任務時，執行完還有順便查詢下有沒有堆積的任務，直到沒有堆積任務了才結束執行緒。
最後一個需要考慮的地方是，上面的程式碼並沒有對任務執行失敗的情況進行處理，也就是說如果某個任務執行失敗了，那麼連重試的機會都沒有。因此，在生產環境使用時，還需要考慮任務處理失敗的情況。有一個簡單的方法是在任務處理時捕獲異常，當在處理過程中出現異常時，就將該任務再放回Redis Sorted中，或者由當前執行緒再重試處理。不過這樣做的話，任務的時效性就不能保證了，有可能本來定在早上7點執行的任務，因為失敗重試的原因，延遲到7點10分才執行完成，這個要根據業務來進行權衡，比如可以在任務描述中增加重試次數或者是否允許重試的欄位，這樣在任務執行失敗時，就能根據不同的任務採取不同的補償措施了。
那麼使用redis實現延時任務有什麼優缺點呢？優點就是可以滿足吞吐量。缺點則是存在任務丟失的風險（當redis例項掛了的時候）。因此，如果對效能要求比較高，同時又能容忍少數情況下任務的丟失，那麼可以使用這種方式來實現。
關注公眾號，檢視更多優質原創文章。
