前言

早在實(shí)習(xí)的時(shí)候，筆者就接觸了spring-task和quartz框架，相信任何java程序員都會(huì)有定時(shí)任務(wù)的需求，在單機(jī)上使用定時(shí)任務(wù)是非常簡(jiǎn)單的，但是在集群環(huán)境中就顯得比較棘手：

• 如何限定只有一臺(tái)機(jī)器在執(zhí)行定時(shí)任務(wù)
• 某臺(tái)服務(wù)宕機(jī)以后如何進(jìn)行故障轉(zhuǎn)移
• 如何確定正在執(zhí)行的是哪一臺(tái)服務(wù)

此問(wèn)題官方肯定有解決方案，資料也不少：quartz集群分布式(并發(fā))部署解決方案-Spring
不過(guò)筆者是個(gè)比較懶的人，上述的解決方案需要配置不少的quartz表，并且業(yè)務(wù)里不需要?jiǎng)討B(tài)配置定時(shí)任務(wù)，對(duì)quartz框架并無(wú)依賴，所以就考慮使用其他方式來(lái)實(shí)現(xiàn)定時(shí)任務(wù)的分布式調(diào)度。

初嘗試Uncode-Schedule

Uncode-Schedule 是筆者在開源中國(guó)中看到的一個(gè)開源的分布式調(diào)度任務(wù)組件，與spring集成比較容易，使用方便，借助了zookeeper。在此我不多做評(píng)價(jià)，畢竟是個(gè)人的一個(gè)開源框架，使用過(guò)程中還是踩了一些坑，并且文檔尚未齊全，用了一段時(shí)間后決定棄用了。有興趣的同學(xué)可以了解下：分布式任務(wù)調(diào)度組件 Uncode-Schedule
所幸在填坑過(guò)程中對(duì)這個(gè)任務(wù)調(diào)度的過(guò)程有了一定了解，于是決定自己實(shí)現(xiàn)一個(gè)簡(jiǎn)單的任務(wù)調(diào)度器。

通過(guò)redis實(shí)現(xiàn)任務(wù)調(diào)度思路

實(shí)現(xiàn)功能之前，回顧下之前遇到的三個(gè)問(wèn)題：?jiǎn)吸c(diǎn)執(zhí)行，故障轉(zhuǎn)移，服務(wù)狀態(tài)。結(jié)合著redis的一些接口特性，解決思路如下：

1. 使用redis作為任務(wù)調(diào)度中心，采用了redis的自動(dòng)過(guò)期與分布式鎖特性
2. 每個(gè)服務(wù)的ip加項(xiàng)目名作為每臺(tái)服務(wù)的唯一別名
3. 通過(guò)redis中對(duì)應(yīng)key值中的value來(lái)判定執(zhí)行的是哪臺(tái)服務(wù): 如redis中key為 schedular_root:projectA, value為192.168.1.187. 意為項(xiàng)目projectA當(dāng)前執(zhí)行任務(wù)的節(jié)點(diǎn)為192.168.1.187這臺(tái)機(jī)器上的服務(wù)
4. 每次執(zhí)行任務(wù)之前判定下redis中schedular_root:projectA是否為空，如果為空，則設(shè)置當(dāng)前ip進(jìn)去，設(shè)置一定時(shí)間的有效期，并執(zhí)行定時(shí)任務(wù)；如果不為空，判斷是否與本機(jī)ip相同，相同則執(zhí)行定時(shí)任務(wù)，否則跳過(guò)
5. 設(shè)置有效期是為了某臺(tái)機(jī)器發(fā)生故障時(shí)能進(jìn)行故障轉(zhuǎn)移

核心流程代碼

此解決方案非常簡(jiǎn)單，核心代碼也十分容易集成，為了減少耦合度，我們采用了spring的aop進(jìn)行實(shí)現(xiàn)。

核心AOP代碼

@Aspect
@Component
@Log4j
public class QuartzAop {
    
    public boolean checkStatus(){
        String key = "schedular_root:projectA";
        try {
            // 這個(gè)接口必然是并發(fā)的，所以加分布式鎖
            while (true) {
                // 一秒的超時(shí)時(shí)間
                boolean lock = RedisUtil.checkLock(key,1);
                if (lock) {
                    // 獲取到鎖，才能跳出
                    break;
                }
            }
            String ip = InetAddress.getLocalHost().getHostAddress();
            // 獲取服務(wù)器上的工作ip
            String currentIp = RedisUtil.get(key);
            // 如果為空的時(shí)候，設(shè)置進(jìn)去
            if(currentIp == null){
                RedisUtil.setex(key, ip, 10);
                return true;
            }
            // 就是當(dāng)前機(jī)器，則返回true
            if(currentIp.equals(ip)){
                return true;
            }else{
                return false;
            }
        } catch (Exception e) {
            log.error(e);
            return false;
        } finally {
            RedisUtil.unLock(key);
        }
    }
    
    @Around("@annotation(org.springframework.scheduling.annotation.Scheduled)")
    public void around(ProceedingJoinPoint jp) throws Throwable{
        if(checkStatus()){
            String ip = InetAddress.getLocalHost().getHostAddress();
            log.info("現(xiàn)在正在執(zhí)行"+jp.getSignature()+":"+ip);
            jp.proceed();
        }
    }
    
}

RedisUtil中的鎖代碼

public static boolean checkLock(String key,int second) {
  
  String lockKey = "lock:" + key;
  try {
   // 1表示之前不存在，設(shè)置成功
   if (setnx(lockKey, "lock") == 1) {
    // 設(shè)置有限期
    setExpiredTime(lockKey, second);
    return true;
   } else {
    // 50毫秒的延遲，避免過(guò)多請(qǐng)求
    try {
     Thread.sleep(50L);
    } catch (InterruptedException e) {
     log.error(e);
    }
    return false;
   }

  } catch (RedisException e) {
   log.error(e);
   return true;
  }
 }

核心代碼已經(jīng)全部放送，關(guān)于redis實(shí)現(xiàn)分布式鎖，會(huì)在另外找個(gè)時(shí)間和大家分享，歡迎關(guān)注～