基於Redis實現分佈式定時任務調度

 

項目開發過程當中，不免會有許多定時任務的需求進來。若是項目中尚未引入quarzt框架的狀況下，咱們一般會使用Spring的@Schedule(cron="* * * * *")註解

樣例以下：

package com.slowcity.redis;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;

public class SentMailTask {
    private static final Logger log = LoggerFactory.getLogger(SentMailTask.class);
   /**
    * 定時任務
    */
    @Scheduled(cron = "0 0/1 * * * ? ")
    public void closeOrderTaskV1() {
        log.info(".........schedule task start.........");
        
        sentMailToCustomer();
        
        log.info(".........schedule task end.........");
    }
     
    public void sentMailToCustomer() {
        log.info(".........sent mail to customer.........");
    }
}
 

這樣實現天然是沒有什麼問題，對於單臺機器部署，任務每一分鐘執行一次。部署多臺機器時，同一個任務會執行屢次

在咱們的項目當中，使用定時任務是避免不了的，咱們在部署定時任務時，一般只部署一臺機器，此時可用性又沒法保證現實狀況是獨立的應用服務一般會部署在兩臺及以上機器的時候，假若有3臺機器，則會出現同一時間3臺機器都會觸發的狀況，結果就是會向客戶發送三封如出一轍的郵件，真讓人頭疼。若是使用quarzt，就不存在這個狀況了。

這種併發的問題，簡單點說是鎖的問題，具體點是分佈式鎖的問題，因此在這段代碼上加個分佈式鎖就能夠了。分佈式鎖，首先想到的是redis,畢竟輪子都是現成的。

package com.slowcity.redis;

import java.util.Collections;
import redis.clients.jedis.Jedis;

public class RedisPool {
    private static final String LOCK_SUCCESS="OK";
    private static final String SET_IF_NOT_EXIST="NX";
    private static final String SET_WITH_EXPIRE_TIME="PX";
    private static final Long RELEASE_SUCCESS=1L;
    
    /**
     * 獲取分佈式鎖
     * @param jedis
     * @param lockKey
     * @param requestID
     * @param expireTime
     * @return
     */
    public static boolean getDistributedLock(Jedis jedis,String lockKey,String requestId,int expireTime) {
        String result = jedis.set(lockKey,requestId,SET_IF_NOT_EXIST,SET_WITH_EXPIRE_TIME,expireTime);
        if(LOCK_SUCCESS.equals(result)) {
            return true;
        }
        return false;
        
    }
    /**
     * 釋放分佈式鎖
     * @param jedis
     * @param lockKey
     * @param requestId
     * @return
     */
    public static boolean releaseDistributedLock(Jedis jedis,String lockKey,String requestId) {
        String script = "if redis.call('get',KEYS[1])== ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
        Object result = jedis.eval(script,Collections.singletonList(lockKey),Collections.singletonList(requestId));
        if(RELEASE_SUCCESS.equals(result)) {
            return true;
        }
        return false;
    }
}

改造一下定時任務，增長分佈式鎖

package com.slowcity.redis;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;

import redis.clients.jedis.Jedis;

public class SentMailTask {
    private static final Logger log = LoggerFactory.getLogger(SentMailTask.class);
   /**
    * 定時任務
    */
    @Scheduled(cron = "0 0/1 * * * ? ")
    public void closeOrderTaskV1() {
        log.info(".........schedule task start.........");
        Jedis jedis = new Jedis("10.2.1.17",6379);
        boolean locked = RedisPool.getDistributedLock(jedis, "", "", 10*1000);
        if(locked) {
            sentMailToCustomer();
        }
        RedisPool.releaseDistributedLock(jedis, "", "");
        jedis.close();
        log.info(".........schedule task end.........");
    }
     
    public void sentMailToCustomer() {
        log.info(".........sent mail to customer.........");
    }
}
 

再執行定時任務，多臺機器部署，只執行一次。

關於jedis對象的獲取，通常都是springboot自動化配置的，全部會想到工廠方法。優化以下：

package com.slowcity.redis;

import java.lang.reflect.Field;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.jedis.JedisConnection;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.util.ReflectionUtils;
import redis.clients.jedis.Jedis;

public class SentMailTask {
    private static final Logger log = LoggerFactory.getLogger(SentMailTask.class);
   
    @Autowired
    private RedisConnectionFactory redisConectionFactory;
    
    /**
    * 定時任務
    */
    @Scheduled(cron = "0 0/1 * * * ? ")
    public void closeOrderTaskV1() {
        log.info(".........schedule task start.........");
        
        RedisConnection redisConnection = redisConectionFactory.getConnection();
        Field jedisField = ReflectionUtils.findField(JedisConnection.class, "jedis");
        Jedis jedis = (Jedis) ReflectionUtils.getField(jedisField, redisConnection);
       
        boolean locked = RedisPool.getDistributedLock(jedis, "lockKey", "requestId", 10*1000);
        if(locked) {
            sentMailToCustomer();
        }
        RedisPool.releaseDistributedLock(jedis, "", "");
        jedis.close();
        log.info(".........schedule task end.........");
    }
     
    public void sentMailToCustomer() {
        log.info(".........sent mail to customer.........");
    }
}
  

不再用擔憂，應用服務多臺機器部署，每臺機器都觸發的尷尬了。若是定時任務不少，最好的仍是老老實實寫個任務調度中心，一來方便管理，二來方便維護。

補充部分：

 一些關於lua腳本的解釋

String script = "if redis.call('get',KEYS[1])== ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
Object result = jedis.eval(script,Collections.singletonList(lockKey),Collections.singletonList(requestId));        

若是一個請求更新緩存的時間比較長，甚至比鎖的有效期還要長，致使在緩存更新的過程當中，鎖就失效了，此時另外一個請求就會獲取鎖，但前一個請求在緩存更新完畢的時候，若是不加以判斷就直接刪除鎖，就會出現誤刪除其它請求建立的鎖的狀況。

【end】

 

一點補充的話，寫完這篇博客後來看其餘博客，也有一種redis鎖是關聯主機ip的，思路上是可行的，不失一個方法點，主要描述以下：

每一個定時任務都在Redis中設置一個Key-Value，Key爲自定義的每一個定時任務的名字（如task1:redis:lock），Value爲服務器Ip，同時設置合適的過時時間（例如設置爲5min）。

每一個節點在執行時，都要進行如下操做：

• 1.是否存在Key，若不存在，則設置Key-Value，Value爲當前節點的IP
• 2.若存在Key，則比較Value是不是當前Ip，如果則繼續執行定時任務，若不是，則不往下執行。