快捷導(dǎo)航

spring-data-redis自定義實現(xiàn)看門狗機(jī)制

更新時間：2024年03月04日 08:53:19 作者：皮卡沖撞

redission看門狗機(jī)制是解決分布式鎖的續(xù)約問題,本文主要介紹了spring-data-redis自定義實現(xiàn)看門狗機(jī)制,具有一定的參考價值,感興趣的可以了解一下

前言

項目中使用redis分布式鎖解決了點贊和樓層排序得問題，所以這里就對這個redis得分布式鎖進(jìn)行了學(xué)習(xí)，一般使用得是redission提供得分布式鎖解決得這個問題，但是知其然更要知其所以然，所以自己就去找了一些資料以及也實踐了一下就此記錄分享一下。

redission分布式鎖看門狗機(jī)制簡單流程圖

在這里插入圖片描述

spring-data-redis實現(xiàn)看門狗機(jī)制指南開始

引入依賴

        <!--redis的依賴-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <!--json工具包-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.60</version>
        </dependency>

配置redis連接以及基礎(chǔ)配置

spring:
  redis:
    host: localhost
    port: 6379
    lettuce:
      timeout: 200000 
    database: 1

在Spring Boot的配置類中創(chuàng)建一個RedisTemplate的Bean:

@Configuration
public class RedisConfig {


    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        // 我們?yōu)榱俗约洪_發(fā)方便，一般直接使用 <String, Object>
        RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
        template.setConnectionFactory(connectionFactory);
        // Json序列化配置
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // String 的序列化
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        // key采用String的序列化方式
        template.setKeySerializer(stringRedisSerializer);
        // hash的key也采用String的序列化方式
        template.setHashKeySerializer(stringRedisSerializer);
        // value序列化方式采用jackson
        template.setValueSerializer(jackson2JsonRedisSerializer);
        // hash的value序列化方式采用jackson
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();
        return template;
    }
}

實現(xiàn)redis分布式鎖工具類

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class RedisLockUtils {
    @Resource
    private RedisTemplate redisTemplate;
    private volatile static Map<String, LockInfo> lockInfoMap = new ConcurrentHashMap<>();
    private static final Long SUCCESS = 1L;
    public static class LockInfo {
        private String key;
        private String value;
        private int expireTime;
        //更新時間
        private long renewalTime;
        //更新間隔
        private long renewalInterval;
        public static LockInfo getLockInfo(String key, String value, int expireTime) {
            LockInfo lockInfo = new LockInfo();
            lockInfo.setKey(key);
            lockInfo.setValue(value);
            lockInfo.setExpireTime(expireTime);
            lockInfo.setRenewalTime(System.currentTimeMillis());
            lockInfo.setRenewalInterval(expireTime*1000 *2 / 3);
            return lockInfo;
        }

        public String getKey() {
            return key;
        }
        public void setKey(String key) {
            this.key = key;
        }
        public String getValue() {
            return value;
        }
        public void setValue(String value) {
            this.value = value;
        }
        public int getExpireTime() {
            return expireTime;
        }
        public void setExpireTime(int expireTime) {
            this.expireTime = expireTime;
        }
        public long getRenewalTime() {
            return renewalTime;
        }
        public void setRenewalTime(long renewalTime) {
            this.renewalTime = renewalTime;
        }
        public long getRenewalInterval() {
            return renewalInterval;
        }
        public void setRenewalInterval(long renewalInterval) {
            this.renewalInterval = renewalInterval;
        }
    }

    /**
     * 使用lua腳本更新redis鎖的過期時間
     * @param lockKey
     * @param value
     * @return 成功返回true, 失敗返回false
     */
    public boolean renewal(String lockKey, String value, int expireTime) {
        String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('expire', KEYS[1], ARGV[2]) else return 0 end";
        DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>();
        redisScript.setResultType(Boolean.class);
        redisScript.setScriptText(luaScript);
        List<String> keys = new ArrayList<>();
        keys.add(lockKey);

        Object result = redisTemplate.execute(redisScript, keys, value, expireTime);
        log.info("更新redis鎖的過期時間：{}", result);
        return (boolean) result;
    }

    /**
     * @param lockKey    鎖
     * @param value      身份標(biāo)識（保證鎖不會被其他人釋放）
     * @param expireTime 鎖的過期時間（單位：秒）
     * @return 成功返回true, 失敗返回false
     */
    public boolean lock(String lockKey, String value, int expireTime) {
        Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent(lockKey, value, expireTime, TimeUnit.SECONDS);
        if(aBoolean){
            lockInfoMap.put(String.valueOf(Thread.currentThread().getId()),LockInfo.getLockInfo(lockKey,value,expireTime));
        }
        return aBoolean;
    }

    /**
     * redisTemplate解鎖
     * @param key
     * @param value
     * @return 成功返回true, 失敗返回false
     */
    public boolean unlock2(String key, String value) {
        Object currentValue = redisTemplate.opsForValue().get(key);
        boolean result = false;
        if (StringUtils.isNotEmpty(String.valueOf(currentValue)) && currentValue.equals(value)) {
            lockInfoMap.remove(String.valueOf(Thread.currentThread().getId()));
            result = redisTemplate.opsForValue().getOperations().delete(key);
        }
        return result;
    }

    /**
     * 定時去檢查redis鎖的過期時間
     */
    @Scheduled(fixedRate = 1000)
    @Async("redisExecutor")
    public void renewal() {
        long now = System.currentTimeMillis();
        for (Map.Entry<String, LockInfo> lockInfoEntry : lockInfoMap.entrySet()) {
            LockInfo lockInfo = lockInfoEntry.getValue();
            System.out.println("++"+lockInfo.key);
            if (lockInfo.getRenewalTime() + lockInfo.getRenewalInterval() < now) {
                renewal(lockInfo.getKey(), lockInfo.getValue(), lockInfo.getExpireTime());
                lockInfo.setRenewalTime(now);
                log.info("lockInfo {}", JSON.toJSONString(lockInfo));
            }
        }
    }

    /**
     * 分布式鎖設(shè)置單獨線程池
     * @return
     */
    @Bean("redisExecutor")
    public ThreadPoolTaskExecutor redisExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(1);
        executor.setMaxPoolSize(1);
        executor.setQueueCapacity(1);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix("redis-renewal-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
        executor.initialize();
        return executor;
    }
}

這個類是一個用于實現(xiàn)分布式鎖的工具類，主要提供了以下功能：

renewal() 方法：定時檢查并更新 Redis 鎖的過期時間。該方法使用 @Scheduled 注解進(jìn)行定時執(zhí)行，通過遍歷 lockInfoMap 中保存的鎖信息，判斷是否需要更新鎖的過期時間，并調(diào)用 renewal() 方法進(jìn)行更新。
renewal(String lockKey, String value, int expireTime) 方法：使用 Lua 腳本更新 Redis 鎖的過期時間。該方法首先定義了一個 Lua 腳本，然后使用 redisTemplate.execute() 方法執(zhí)行該腳本，并傳入相應(yīng)的參數(shù)。如果執(zhí)行成功，則返回 true，否則返回 false。
lock(String lockKey, String value, int expireTime) 方法：獲取分布式鎖。該方法使用 Redis 的 setIfAbsent() 方法嘗試將鎖的鍵值對存儲到 Redis 中，并設(shè)置相應(yīng)的過期時間。如果存儲成功，則返回 true，表示獲取鎖成功；否則返回 false，表示獲取鎖失敗。
unlock2(String key, String value) 方法：釋放分布式鎖。該方法首先獲取 Redis 中當(dāng)前的鎖值，然后判斷鎖值是否和傳入的 value 相等。如果相等，則從 lockInfoMap 中移除鎖信息，并調(diào)用 Redis 的 delete() 方法刪除鎖的鍵值對。最后返回刪除結(jié)果，表示是否成功釋放鎖。
redisExecutor() 方法：配置一個單獨的線程池用于執(zhí)行 renewal() 方法。該方法創(chuàng)建一個 ThreadPoolTaskExecutor 對象，并設(shè)置相關(guān)的屬性，如核心線程數(shù)、最大線程數(shù)、隊列容量等。

直接失敗和鎖重試機(jī)制實現(xiàn)

直接失敗的方式，就是調(diào)用獲取鎖的方法判斷是否加鎖成功，失敗則直接中斷方法執(zhí)行返回

    @GetMapping("/test2")
    public Object test2(){
        boolean a = redisLockUtils.lock("A", "111", 5);
        if(!a){
            return "獲取鎖失敗";
        }
        try{
            System.out.println("執(zhí)行開始-------------------test2");
            TimeUnit.SECONDS.sleep(12);
            System.out.println("執(zhí)行結(jié)束-------------------test2");
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {

            redisLockUtils.unlock2("A","111");
            System.out.println("釋放鎖-----------------------test2");
        }
        return null;
   }

鎖重試，這里是封裝了獲取鎖的方法，加入了一個重試次數(shù)的限制，通過使用while循環(huán)去嘗試獲取鎖。

private Boolean getLock(int tryNum, String key, String value, int exp) throws InterruptedException {
    int i = 0; // 初始化計數(shù)器，記錄嘗試獲取鎖的次數(shù)
    boolean flag = false; // 初始化標(biāo)志變量，表示獲取鎖的結(jié)果

    while (true) { // 循環(huán)進(jìn)行嘗試獲取鎖的操作
        if (i == tryNum) { // 判斷是否達(dá)到嘗試獲取鎖的最大次數(shù)
            return flag; // 返回當(dāng)前的獲取鎖結(jié)果
        }

        flag = redisLockUtils.lock(key, value, exp); // 嘗試獲取鎖，返回是否成功獲取到鎖的結(jié)果

        if (flag) { // 如果成功獲取到鎖
            return flag; // 直接返回獲取鎖結(jié)果為 true
        } else { // 如果未能成功獲取到鎖
            i++; // 計數(shù)器加一，表示已經(jīng)嘗試了一次獲取鎖的操作
            TimeUnit.SECONDS.sleep(1); // 暫停一秒鐘，等待一段時間后再進(jìn)行下一次獲取鎖的嘗試
        }
    }
}