Redis分布式锁实现的方法是什么

其他教程   发布日期:2024年11月22日   浏览次数:112

本篇内容主要讲解“Redis分布式锁实现的方法是什么”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“Redis分布式锁实现的方法是什么”吧!

    一、分布式锁是什么

    分布式锁是 满足分布式系统或集群模式下多进程可见并且互斥的锁。

    基于Redis实现分布式锁:

    1、获取锁

    • 互斥:确保只能有一个线程获取锁;

    • 非阻塞:尝试获取锁,成功返回true,失败返回false;

    添加锁过期时间,避免服务宕机引起死锁。

    1. SET lock thread1 NX EX 10

    2、释放锁

    • 手动释放;

      1. DEL key1
    • 超时释放,获取锁时添加一个超时锁;

    二、代码实例

    1. package com.guor.utils;
    2. import org.springframework.data.redis.core.StringRedisTemplate;
    3. import java.util.concurrent.TimeUnit;
    4. public class RedisLock implements ILock{
    5. private String name;
    6. private StringRedisTemplate stringRedisTemplate;
    7. public RedisLock(String name, StringRedisTemplate stringRedisTemplate) {
    8. this.name = name;
    9. this.stringRedisTemplate = stringRedisTemplate;
    10. }
    11. private static final String KEY_PREFIX = "lock:";
    12. @Override
    13. public boolean tryLock(long timeout) {
    14. // 获取线程唯一标识
    15. long threadId = Thread.currentThread().getId();
    16. // 获取锁
    17. Boolean success = stringRedisTemplate.opsForValue()
    18. .setIfAbsent(KEY_PREFIX + name, threadId+"", timeout, TimeUnit.SECONDS);
    19. // 防止拆箱的空指针异常
    20. return Boolean.TRUE.equals(success);
    21. }
    22. @Override
    23. public void unlock() {
    24. stringRedisTemplate.delete(KEY_PREFIX + name);
    25. }
    26. }

    上面代码存在锁误删问题:

    1. 如果线程1获取锁,但线程1发生了阻塞,导致Redis超时释放锁;

    2. 此时,线程2尝试获取锁,成功,并执行业务;

    3. 此时,线程1重新开始执行任务,并执行完毕,执行释放锁(即删除锁);

    4. 但是,线程1删除的锁,和线程2的锁是同一把锁,这就是

      1. 分布式锁误删问题

    在释放锁时,释放线程自己的分布式锁,就可以解决这个问题。

    1. package com.guor.utils;
    2. import cn.hutool.core.lang.UUID;
    3. import org.springframework.data.redis.core.StringRedisTemplate;
    4. import java.util.concurrent.TimeUnit;
    5. public class RedisLock implements ILock{
    6. private String name;
    7. private StringRedisTemplate stringRedisTemplate;
    8. public RedisLock(String name, StringRedisTemplate stringRedisTemplate) {
    9. this.name = name;
    10. this.stringRedisTemplate = stringRedisTemplate;
    11. }
    12. private static final String KEY_PREFIX = "lock:";
    13. private static final String UUID_PREFIX = UUID.randomUUID().toString(true) + "-";
    14. @Override
    15. public boolean tryLock(long timeout) {
    16. // 获取线程唯一标识
    17. String threadId = UUID_PREFIX + Thread.currentThread().getId();
    18. // 获取锁
    19. Boolean success = stringRedisTemplate.opsForValue()
    20. .setIfAbsent(KEY_PREFIX + name, threadId, timeout, TimeUnit.SECONDS);
    21. // 防止拆箱的空指针异常
    22. return Boolean.TRUE.equals(success);
    23. }
    24. @Override
    25. public void unlock() {
    26. // 获取线程唯一标识
    27. String threadId = UUID_PREFIX + Thread.currentThread().getId();
    28. // 获取锁中的标识
    29. String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
    30. // 判断标示是否一致
    31. if(threadId.equals(id)) {
    32. // 释放锁
    33. stringRedisTemplate.delete(KEY_PREFIX + name);
    34. }
    35. }
    36. }

    三、基于

    1. SETNX
    实现的分布式锁存在下面几个问题

    1、不可重入

    同一个线程无法多次获取同一把锁。

    2、不可重试

    获取锁只尝试一次就返回false,没有重试机制。

    3、超时释放

    锁的超时释放虽然可以避免死锁,但如果业务执行耗时较长,也会导致锁释放,存在安全隐患。

    4、主从一致性

    如果Redis是集群部署的,主从同步存在延迟,当主机宕机时,此时会选一个从作为主机,但是此时的从没有锁标识,此时,其它线程可能会获取到锁,导致安全问题。

    四、Redisson实现分布式锁

    Redisson是一个在Redis的基础上实现的Java驻内存数据网格。它不仅提供了一系列的分布式的Java常用对象,还提供了许多分布式服务,其中包含各种分布式锁的实现。

    1、pom

    1. <!--redisson-->
    2. <dependency>
    3. <groupId>org.redisson</groupId>
    4. <artifactId>redisson</artifactId>
    5. <version>3.13.6</version>
    6. </dependency>

    2、配置类

    1. package com.guor.config;
    2. import org.redisson.Redisson;
    3. import org.redisson.api.RedissonClient;
    4. import org.redisson.config.Config;
    5. import org.springframework.context.annotation.Bean;
    6. import org.springframework.context.annotation.Configuration;
    7. @Configuration
    8. public class RedissonConfig {
    9. @Bean
    10. public RedissonClient redissonClient(){
    11. // 配置
    12. Config config = new Config();
    13. /**
    14. * 单点地址useSingleServer,集群地址useClusterServers
    15. */
    16. config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
    17. // 创建RedissonClient对象
    18. return Redisson.create(config);
    19. }
    20. }

    3、测试类

    1. package com.guor;
    2. import lombok.extern.slf4j.Slf4j;
    3. import org.junit.jupiter.api.BeforeEach;
    4. import org.junit.jupiter.api.Test;
    5. import org.redisson.api.RLock;
    6. import org.redisson.api.RedissonClient;
    7. import org.springframework.boot.test.context.SpringBootTest;
    8. import javax.annotation.Resource;
    9. import java.util.concurrent.TimeUnit;
    10. @Slf4j
    11. @SpringBootTest
    12. class RedissonTest {
    13. @Resource
    14. private RedissonClient redissonClient;
    15. private RLock lock;
    16. @BeforeEach
    17. void setUp() {
    18. // 获取指定名称的锁
    19. lock = redissonClient.getLock("nezha");
    20. }
    21. @Test
    22. void test() throws InterruptedException {
    23. // 尝试获取锁
    24. boolean isLock = lock.tryLock(1L, TimeUnit.SECONDS);
    25. if (!isLock) {
    26. log.error("获取锁失败");
    27. return;
    28. }
    29. try {
    30. log.info("哪吒最帅,哈哈哈");
    31. } finally {
    32. // 释放锁
    33. lock.unlock();
    34. }
    35. }
    36. }

    五、探索tryLock源码

    1、tryLock源码

    尝试获取锁

    1. public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
    2. // 最大等待时间
    3. long time = unit.toMillis(waitTime);
    4. long current = System.currentTimeMillis();
    5. long threadId = Thread.currentThread().getId();
    6. Long ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
    7. if (ttl == null) {
    8. return true;
    9. } else {
    10. // 剩余等待时间 = 最大等待时间 - 获取锁失败消耗的时间
    11. time -= System.currentTimeMillis() - current;
    12. if (time <= 0L) {// 获取锁失败
    13. this.acquireFailed(waitTime, unit, threadId);
    14. return false;
    15. } else {
    16. // 再次尝试获取锁
    17. current = System.currentTimeMillis();
    18. // subscribe订阅其它释放锁的信号
    19. RFuture<RedissonLockEntry> subscribeFuture = this.subscribe(threadId);
    20. // 当Future在等待指定时间time内完成时,返回true
    21. if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
    22. if (!subscribeFuture.cancel(false)) {
    23. subscribeFuture.onComplete((res, e) -> {
    24. if (e == null) {
    25. // 取消订阅
    26. this.unsubscribe(subscribeFuture, threadId);
    27. }
    28. });
    29. }
    30. this.acquireFailed(waitTime, unit, threadId);
    31. return false;// 获取锁失败
    32. } else {
    33. try {
    34. // 剩余等待时间 = 剩余等待时间 - 获取锁失败消耗的时间
    35. time -= System.currentTimeMillis() - current;
    36. if (time <= 0L) {
    37. this.acquireFailed(waitTime, unit, threadId);
    38. boolean var20 = false;
    39. return var20;
    40. } else {
    41. boolean var16;
    42. do {
    43. long currentTime = System.currentTimeMillis();
    44. // 重试获取锁
    45. ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
    46. if (ttl == null) {
    47. var16 = true;
    48. return var16;
    49. }
    50. // 再次失败了,再看一下剩余时间
    51. time -= System.currentTimeMillis() - currentTime;
    52. if (time <= 0L) {
    53. this.acquireFailed(waitTime, unit, threadId);
    54. var16 = false;
    55. return var16;
    56. }
    57. // 再重试获取锁
    58. currentTime = System.currentTimeMillis();
    59. if (ttl >= 0L && ttl < time) {
    60. // 通过信号量的方式尝试获取信号,如果等待时间内,依然没有结果,会返回false
    61. ((RedissonLockEntry)subscribeFuture.getNow()).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
    62. } else {
    63. ((RedissonLockEntry)subscribeFuture.getNow()).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
    64. }
    65. time -= System.currentTimeMillis() - currentTime;
    66. } while(time > 0L);
    67. this.acquireFailed(waitTime, unit, threadId);
    68. var16 = false;
    69. return var16;
    70. }
    71. } finally {
    72. this.unsubscribe(subscribeFuture, threadId);
    73. }
    74. }
    75. }
    76. }
    77. }

    2、重置锁的有效期

    1. private void scheduleExpirationRenewal(long threadId) {
    2. RedissonLock.ExpirationEntry entry = new RedissonLock.ExpirationEntry();
    3. // this.getEntryName():锁的名字,一个锁对应一个entry
    4. // putIfAbsent:如果不存在,将锁和entry放到map里
    5. RedissonLock.ExpirationEntry oldEntry = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.putIfAbsent(this.getEntryName(), entry);
    6. if (oldEntry != null) {
    7. // 同一个线程多次获取锁,相当于重入
    8. oldEntry.addThreadId(threadId);
    9. } else {
    10. // 如果是第一次
    11. entry.addThreadId(threadId);
    12. // 更新有效期
    13. this.renewExpiration();
    14. }
    15. }

    更新有效期,递归调用更新有效期,永不过期

    1. private void renewExpiration() {
    2. // 从map中得到当前锁的entry
    3. RedissonLock.ExpirationEntry ee = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());
    4. if (ee != null) {
    5. // 开启延时任务
    6. Timeout task = this.commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
    7. public void run(Timeout timeout) throws Exception {
    8. RedissonLock.ExpirationEntry ent = (RedissonLock.ExpirationEntry)RedissonLock.EXPIRATION_RENEWAL_MAP.get(RedissonLock.this.getEntryName());
    9. if (ent != null) {
    10. // 取出线程id
    11. Long threadId = ent.getFirstThreadId();
    12. if (threadId != null) {
    13. // 刷新有效期
    14. RFuture<Boolean> future = RedissonLock.this.renewExpirationAsync(threadId);
    15. future.onComplete((res, e) -> {
    16. if (e != null) {
    17. RedissonLock.log.error("Can't update lock " + RedissonLock.this.getName() + " expiration", e);
    18. } else {
    19. if (res) {
    20. // 递归调用更新有效期,永不过期
    21. RedissonLock.this.renewExpiration();
    22. }
    23. }
    24. });
    25. }
    26. }
    27. }
    28. }, this.internalLockLeaseTime / 3L, TimeUnit.MILLISECONDS);// 10S
    29. ee.setTimeout(task);
    30. }
    31. }

    更新有效期

    1. protected RFuture<Boolean> renewExpirationAsync(long threadId) {
    2. return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
    3. // 判断当前线程的锁是否是当前线程
    4. "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
    5. // 更新有效期
    6. redis.call('pexpire', KEYS[1], ARGV[1]);
    7. return 1;
    8. end;
    9. return 0;",
    10. Collections.singletonList(this.getName()), this.internalLockLeaseTime, this.getLockName(threadId));
    11. }

    3、调用lua脚本

    1. <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    2. // 锁释放时间
    3. this.internalLockLeaseTime = unit.toMillis(leaseTime);
    4. return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, command,
    5. // 判断锁成功
    6. "if (redis.call('exists', KEYS[1]) == 0) then
    7. redis.call('hincrby', KEYS[1], ARGV[2], 1); // 如果不存在,记录锁标识,次数+1
    8. redis.call('pexpire', KEYS[1], ARGV[1]); // 设置锁有效期
    9. return nil; // 相当于Java的null
    10. end;
    11. if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
    12. redis.call('hincrby', KEYS[1], ARGV[2], 1); // 如果存在,判断锁标识是否是自己的,次数+1
    13. redis.call('pexpire', KEYS[1], ARGV[1]); // 设置锁有效期
    14. return nil;
    15. end;
    16. // 判断锁失败,pttl:指定锁剩余有效期,单位毫秒,KEYS[1]:锁的名称
    17. return redis.call('pttl', KEYS[1]);",
    18. Collections.singletonList(this.getName()), this.internalLockLeaseTime, this.getLockName(threadId));
    19. }

    六、释放锁unlock源码

    1、取消更新任务

    1. public RFuture<Void> unlockAsync(long threadId) {
    2. RPromise<Void> result = new RedissonPromise();
    3. RFuture<Boolean> future = this.unlockInnerAsync(threadId);
    4. future.onComplete((opStatus, e) -> {
    5. // 取消更新任务
    6. this.cancelExpirationRenewal(threadId);
    7. if (e != null) {
    8. result.tryFailure(e);
    9. } else if (opStatus == null) {
    10. IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " + this.id + " thread-id: " + threadId);
    11. result.tryFailure(cause);
    12. } else {
    13. result.trySuccess((Object)null);
    14. }
    15. });
    16. return result;
    17. }

    2、删除定时任务

    1. void cancelExpirationRenewal(Long threadId) {
    2. // 从map中取出当前锁的定时任务entry
    3. RedissonLock.ExpirationEntry task = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());
    4. if (task != null) {
    5. if (threadId != null) {
    6. task.removeThreadId(threadId);
    7. }
    8. // 删除定时任务
    9. if (threadId == null || task.hasNoThreads()) {
    10. Timeout timeout = task.getTimeout();
    11. if (timeout != null) {
    12. timeout.cancel();
    13. }
    14. EXPIRATION_RENEWAL_MAP.remove(this.getEntryName());
    15. }
    16. }
    17. }

    以上就是Redis分布式锁实现的方法是什么的详细内容,更多关于Redis分布式锁实现的方法是什么的资料请关注九品源码其它相关文章!