基于redis的分布式锁的实现，包含任务续约看门狗程序以及注解版分布式锁

落日映苍穹つ 2022-12-09 02:40 88阅读 0赞

# 一、什么是分布式锁 #

分布式锁是不同的应用之间的锁，我们常见的锁有ReentrantLock，synchronized,但是这些只是本地锁，也就是在同一个应用级别的锁，如果我们的服务是一个分布式服务，而且要在不同的应用之间加锁，那么本地锁就无法实现这样的功能，只能采用分布式锁才能实现。

### 二、分布式锁的实现方案 ###

目前公认的分布式锁解决方案有基于数据库、redis、zk的分布式锁，但是基于数据库的锁性能不是很好，基于zk的分布式锁需要加入zk但是我们现有的服务不需要zk，我们已经有了redis的环境，所以选择了redis的分布式锁。

### 三、redis分布式锁实现 ###

（1）接口

interface DistributedLock {
        fun tryLock(): Boolean
        fun lock()
        fun unLock()
    }

（2）分布式锁实现

class RedisDistributedLock : DistributedLock {
    
        private var stringRedisTemplate: StringRedisTemplate
        private var rKey: String
        private var rValue: String
        private var leaseTime: Long = 6000L
    
        private var executor: ExecutorService = Executors.newCachedThreadPool(CustomizableThreadFactory("redis-lock-registry-"))
    
        private var executorWatchDog: ScheduledExecutorService = Executors.newScheduledThreadPool(10)
    
        private var localLock = ReentrantLock()
        private var lockedAt: Long = 0L
    
        @Volatile
        private var unlinkAvailable = true
    
        private var enableWatchDog: Boolean = false
        private var threshold: Long = 2000L
        private var leaseRenewalTime: Long = 1000L
        private var leaseRenewalNumber: Int = 5
    
        constructor(stringRedisTemplate: StringRedisTemplate, rKey: String, rValue: String, leaseTime: Long, enableWatchDog: Boolean, threshold: Long, leaseRenewalTime: Long, leaseRenewalNumber: Int) {
            this.stringRedisTemplate = stringRedisTemplate
            this.rKey = "RedisDistributedLock:$rKey"
            this.rValue = rValue
            this.leaseTime = leaseTime
            this.enableWatchDog = enableWatchDog
            this.threshold = threshold
            this.leaseRenewalTime = leaseRenewalTime
            this.leaseRenewalNumber = leaseRenewalNumber
        }
    
        constructor(stringRedisTemplate: StringRedisTemplate, rKey: String, rValue: String) : this(stringRedisTemplate, rKey, rValue, 6000L, false, 0, 0, 0)
    
        override fun tryLock(): Boolean {
            return try {
                tryLock(0, TimeUnit.MILLISECONDS)
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
                false
            }
        }
    
        fun tryLock(time: Long, unit: TimeUnit): Boolean {
            val now = System.currentTimeMillis()
            if (!this.localLock.tryLock(time, unit)) {
                return false
            }
            try {
                val expire = now + TimeUnit.MILLISECONDS.convert(time, unit)
                var acquired: Boolean
                while (redisLock().also { acquired = it }.not() && System.currentTimeMillis() < expire) {
                    Thread.sleep(100)
                }
                if (!acquired) {
                    this.localLock.unlock()
                }
                return acquired
            } catch (e: Exception) {
                this.localLock.unlock()
                rethrowAsLockException(e)
            }
            return false
        }
    
        override fun lock() {
            localLock.lock()
            while (true) {
                try {
                    while (!redisLock()) {
                        Thread.sleep(100)
                    }
                    break
                } catch (e: InterruptedException) {
                } catch (e: Exception) {
                    localLock.unlock()
                    rethrowAsLockException(e)
                }
            }
    
            if (enableWatchDog) {
                val times = AtomicInteger(0)
                val currentThread = Thread.currentThread()
                this.executorWatchDog.scheduleAtFixedRate({
                    if (times.get() > this.leaseRenewalNumber) {
                        // This is not to throw a real exception, just to make the log print good-looking
                        RuntimeException("RedisDistributedLock's value is ${this.rValue}, leaseRenewalNumber Threshold exceeded in RedisDistributedLock.").printStackTrace()
                        // if main thread is sleep, wait or blocking,main thread can interrupt
                        currentThread.interrupt()
                        throw RuntimeException("exit watchDog.")
                    }
                    watchDog(threshold, leaseRenewalTime, times)
                }, 0, 100L, TimeUnit.MILLISECONDS)
            }
        }
    
        private fun redisLock(): Boolean {
            val result = this.stringRedisTemplate.opsForValue().setIfAbsent(
                    rKey,
                    rValue,
                    leaseTime,
                    TimeUnit.MILLISECONDS
            ) ?: false
    
            if (result) {
                this.lockedAt = System.currentTimeMillis()
            }
            return result
        }
    
        override fun unLock() {
            check(localLock.isHeldByCurrentThread) { "You do not own lock at " + this.rKey }
            if (localLock.holdCount > 1) {
                localLock.unlock()
                return
            }
            try {
                check(isAcquiredInThisProcess()) {
                    "Lock was released in the store due to expiration. " +
                            "The integrity of data protected by this lock may have been compromised."
                }
                if (Thread.currentThread().isInterrupted) {
                    this.executor.execute { this.removeLockKey() }
                } else {
                    removeLockKey()
                }
            } catch (e: Exception) {
                ReflectionUtils.rethrowRuntimeException(e)
            } finally {
                localLock.unlock()
            }
        }
    
        private fun isAcquiredInThisProcess(): Boolean {
            return this.rValue ==
                    this.stringRedisTemplate.boundValueOps(this.rKey).get()
        }
    
        private fun removeLockKey() {
            if (this.unlinkAvailable) {
                try {
                    this.stringRedisTemplate.unlink(this.rKey)
                } catch (ex: Exception) {
                    this.unlinkAvailable = false
                    this.stringRedisTemplate.delete(this.rKey)
                }
            } else {
                this.stringRedisTemplate.delete(this.rKey)
            }
        }
    
        private fun rethrowAsLockException(e: Exception) {
            throw CannotAcquireLockException("Failed to lock mutex at " + this.rKey, e)
        }
    
        private fun watchDog(threshold: Long, leaseRenewalTime: Long, times: AtomicInteger) {
            val expire = this.stringRedisTemplate.getExpire(this.rKey) * 1000
            if (expire > 0 && isAcquiredInThisProcess()) {
                if (expire <= threshold) {
                    times.incrementAndGet()
                    this.stringRedisTemplate.expire(this.rKey, Duration.ofMillis(expire + leaseRenewalTime))
                }
            } else {
                throw RuntimeException("exit watchDog.")
            }
        }
    }

其他的地方与普通的redis分布式锁差别不大，都是使用了setnx来获取锁，不同的是，我使用了一个线程池，来周期的去便利，查看锁的过期时间有没有到一个阈值，如果锁的过期时间到达阈值，并且开启了续约机制，则进行续约，但是续约不能无限制续约，设置了一个最大续约次数，只要到达了这个最大次数，不管有没有完成任务都会结束续约线程。

如果是阻塞、等待、睡眠等状态，续约线程还会调用interrupt去中断任务线程，但是如果是循环次数过长或者死循环，则无法中断任务线程，会打印一个异常日志，然后释放锁。

### 四、注解的实现 ###

注解的话，采用Spring的aop，就不过多赘述了，直接贴代码。

@Target(AnnotationTarget.FUNCTION)
    @Retention(AnnotationRetention.RUNTIME)
    @Documented
    @Inherited
    annotation class RedisDistributedLock(
        /**
         * 方法上第一个参数的某一个字段
         */
        val rKeyParam: String = "",
        /**
         * 方法上第一个参数的某一个字段
         */
        val rValueParam: String = "",
        val leaseTime: Long = 5000L,
        val leaseRenewTime: Long = 5000L,
        val threshold: Long = 2000L,
        val leaseRenewalNumber: Int = 3,
        val enableWatchDog: Boolean = false
    )

@Aspect
    class RedisLockSupport {
    
        @Autowired
        private lateinit var beanFactory: ConfigurableListableBeanFactory
    
        @Autowired
        private lateinit var redisLockProperty: RedisLockProperty
    
        @Pointcut("@annotation(RedisDistributedLock)")
        fun redisLockPointCut(redisDistributedLock: RedisDistributedLock) {
        }
    
        @Around("redisLockPointCut(redisDistributedLock)")
        fun around(joinPoint: ProceedingJoinPoint, redisDistributedLock: RedisDistributedLock): Any? {
            val args = joinPoint.args
            val key: String
            val value: String
            if (args == null || args.isEmpty()) {
                key = joinPoint.signature.name
                value = UUID.randomUUID().toString() + System.currentTimeMillis()
            } else {
                key = Json.deserialize(args[0].toJson()).get(redisDistributedLock.rKeyParam).textValue()
                value = Json.deserialize(args[0].toJson()).get(redisDistributedLock.rValueParam).textValue()
            }
            var stringRedisTemplate: StringRedisTemplate? = null
            val beanNamesForType = beanFactory.getBeanNamesForType(StringRedisTemplate::class.java)
            loop@ for (it in beanNamesForType) {
                val abstractBeanDefinition = beanFactory.getBeanDefinition(it) as AbstractBeanDefinition
                for (qualifier in abstractBeanDefinition.qualifiers) {
                    if (qualifier.typeName == redisLockProperty.redisQualifier) {
                        stringRedisTemplate = beanFactory.getBean(it) as StringRedisTemplate
                        break@loop
                    }
                }
            }
            if (stringRedisTemplate == null) {
                throw NullPointerException("stringRedisTemplate is not be null.")
            }
            val redisDistributedLock = RedisDistributedLock(
                    stringRedisTemplate,
                    key,
                    value,
                    RedisDistributedLock.leaseTime,
                    RedisDistributedLock.enableWatchDog,
                    RedisDistributedLock.threshold,
                    RedisDistributedLock.leaseRenewTime,
                    RedisDistributedLock.leaseRenewalNumber
                    )
            if (redisDistributedLock.tryLock()) {
                try {
                    return joinPoint.proceed()
                } finally {
                    redisDistributedLock.unLock()
                }
            } else {
                throw RuntimeException("locked failed.")
            }
        }
    }