分布式定時任務重復執行解決方案--redis篇

問題：定時任務在只部署一台服務器時沒有問題，當需要集群時，就會重復執行多次。

解決方案：1. 利用數據庫樂觀鎖；2. 基於Redis的分布式鎖；3. 基於ZooKeeper的分布式鎖。

這里我使用的是redis分布鎖的方式實現，自己封裝了一個注解，如有問題請聯系我一下，謝謝！

加鎖 :同一個定時任務同時多次給redis加鎖(key),如果存在key,則加鎖失敗,如果不存在,則嘗試去加鎖,返回加鎖結果。 

解鎖: 設置一下過期時間為20秒(可根據任務執行長短調整),過期后自動釋放掉，當定時任務執行完后redis還沒有過期是就手動解鎖。

封裝aop注解：

package com.demo.aop;

 4 import java.lang.annotation.*;

/**
 * ************************************************
 * 功能描述:   TODO
 *
 * @author shuangping.yang
 * @version 1.0
 * @ClassName RedisTryLock
 * @date 2020.07.15 下午 03:53 創建文件
 * @see ************************************************
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RedisTryLock {
    /**
     * 鎖的有效時間長，單位：秒
     *
     * @return
     */
    int expireTime() default 10;

    /**
     * 自定義鎖的keyName（不用包含namespace，內部已實現）
     */
31     String keyName() default "";
}

 

核心代碼實現：

package com.demo.aop;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.concurrent.TimeUnit;

/**
 * @author shuangping.yang
 */
@Slf4j
@Aspect
@Component
public class RedisTryLockAspect {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    InetAddress addr = null;


    @Around("execution(* *.*(..)) && @annotation(com.demo.aop.RedisTryLock)")
    public void redisTryLockPoint(ProceedingJoinPoint pjp) throws Exception {
        String defKey = "redis:lock:";
        RedisTryLock annotation = null;
        Method method = null;
        //獲得所在切點的該類的class對象
        Class aClass = pjp.getTarget().getClass();
        //獲取該切點所在方法的名稱,每一個使用切面的方法
        String name = pjp.getSignature().getName();
        try {
            //通過反射獲得該方法
            method = aClass.getMethod(name);
            //獲得該注解
            annotation = method.getAnnotation(RedisTryLock.class);
            //獲取注解對應的值keyName，expireTime
            String keyName = annotation.keyName();
            Integer expireTime = annotation.expireTime();
            Assert.isTrue(0 != expireTime, "redis lock's expireTime is null");
            //獲取本機ip
            try {
                addr = InetAddress.getLocalHost();

            } catch (UnknownHostException e) {
                log.error("獲取IP失敗--》", e);

            }
            String ip = addr.getHostAddress();
            // 設置redis key值

            defKey = StringUtils.isBlank(keyName) ? defKey + aClass.getDeclaringClass() + method.getName() : defKey + keyName;
            //根據redis 鎖的原理判斷是否執行成功，設值成功說明其他服務器沒有執行定時任務，反則正在執行
            if (redisTemplate.opsForValue().setIfAbsent(defKey, ip)) {
                redisTemplate.expire(defKey, expireTime, TimeUnit.SECONDS);
                log.info("獲得分布式鎖成功! key:{}", defKey);
                pjp.proceed();
                redisTemplate.delete(defKey);
                log.info("定時任務執行完，釋放分布式鎖成功,key:{}", defKey);
                return;

            }
            Object redisVal = redisTemplate.opsForValue().get(defKey);
            log.info("{}已在{}機器上占用分布式鎖，聚類任務正在執行", defKey, redisVal);

        } catch (NoSuchMethodException e) {
            e.printStackTrace();
            log.error("Facet aop failed error {}", e.getLocalizedMessage());

        } catch (Throwable throwable) {
            throwable.printStackTrace();

        }

    }
}

 

 

使用示例：

    @Scheduled(cron = "0 0 0/1 * * ?")
    @RedisTryLock(keyName = "repeat_flow_direction_file_task", expireTime = 180)
    public void redisTask() {
        //業務代碼實現
    }

希望能幫助到大家，謝謝！