关于redis:加锁了还有并发问题Redis分布式锁你真的了解

41次阅读

共计 14815 个字符,预计需要花费 38 分钟才能阅读完成。

新接手的我的项目,偶然会呈现账不平的问题。之前的技术老大临走时给的解释是:排查了,没找到起因,之后太忙就没再解决,可能是框架的起因……

既然我的项目交付到手中,这样的问题是必须要解决的。梳理了所有账务解决逻辑,最终找到了起因:数据库并发操作热点账户导致。就这这个问题,来聊一聊分布式系统下基于 Redis 的分布式锁。顺便也合成一下问题造成起因及解决方案。

起因剖析

零碎并发量并不高,存在热点账户,但也不至于那么重大。问题的本源在于零碎架构设计,人为的制作了并发。场景是这样的:商户批量导入一批数据,零碎会进行前置解决,并对账户余额进行增减。

此时,另外一个定时工作,也会对账户进行扫描更新。而且对同一账户的操作散布到各个系统当中,热点账户也就呈现了。

针对此问题的解决方案,从架构层面能够思考将账务零碎进行抽离,集中在一个零碎中进行解决,所有的数据库事务及执行程序由账务零碎来兼顾解决。从技术方面来讲,则能够通过锁机制来对热点账户进行加锁。

本篇文章就针对热点账户基于分布式锁的实现形式进行具体的解说。

锁的剖析

在 Java 的多线程环境下,通常有几类锁能够应用:

  • JVM 内存模型级别的锁,罕用的有:synchronized、Lock 等;
  • 数据库锁,比方乐观锁,乐观锁等;
  • 分布式锁;

JVM 内存级别的锁,能够保障单体服务下线程的安全性,比方多个线程拜访 / 批改一个全局变量。但当零碎进行集群部署时,JVM 级别的本地锁就无能为力了。

乐观锁与乐观锁

像上述案例中,热点账户就属于分布式系统中的共享资源,咱们通常会采纳 数据库锁 分布式锁 来进行解决。

数据库锁,又分为 乐观锁 乐观锁

乐观锁 是基于数据库(Mysql 的 InnoDB)提供的排他锁来实现的。在进行事务操作时,通过 select … for update 语句,MySQL 会对查问后果集中每行数据都增加排他锁,其余线程对该记录的更新与删除操作都会阻塞。从而达到共享资源的程序执行(批改);

乐观锁 是绝对乐观锁而言的,乐观锁假如数据个别状况不会造成抵触,所以在数据进行提交更新的时候,才会正式对数据的抵触与否进行检测。如果抵触则返回给用户异样信息,让用户决定如何去做。乐观锁实用于读多写少的场景,这样能够进步程序的吞吐量。在乐观锁实现时通常会基于记录状态或增加 version 版本来进行实现。

乐观锁生效场景

我的项目中应用了乐观锁,但乐观锁却生效了。这也是应用乐观锁时,常见的误区,上面来剖析一下。

失常应用乐观锁的流程:

  • 通过 select … for update 锁定记录;
  • 计算新余额,批改金额并存储;
  • 执行实现开释锁;

常常犯错的解决流程:

  • 查问账户余额,计算新余额;
  • 通过 select … for update 锁定记录;
  • 批改金额并存储;
  • 执行实现开释锁;

谬误的流程中,比方 A 和 B 服务查问到的余额都是 100,A 扣减 50,B 扣减 40,而后 A 锁定记录,更新数据库为 50;A 开释锁之后,B 锁定记录,更新数据库为 60。显然,后者把前者的更新给笼罩掉了。解决的计划就是扩充锁的范畴,将锁提前到计算新余额之前。

通常乐观锁对数据库的压力是十分大的,在实践中通常会依据场景应用乐观锁或分布式锁等形式来实现。

上面进入正题,讲讲基于 Redis 的分布式锁实现。

Redis 分布式锁实战演习

这里以 Spring Boot、Redis、Lua 脚本为例来演示分布式锁的实现。为了简化解决,示例中 Redis 既承当了分布式锁的性能,也承当了数据库的性能。

场景构建

集群环境下,对同一个账户的金额进行操作,根本步骤:

  • 从数据库读取用户金额;
  • 程序修改金额;
  • 再将最新金额存储到数据库;

上面从最后不加锁,不同步解决,逐渐推演出最终的分布式锁。

根底集成及类构建

筹备一个不加锁解决的根底业务环境。

首先在 Spring Boot 我的项目中引入相干依赖:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

账户对应实体类 UserAccount:

public class UserAccount {

    // 用户 ID
    private String userId;
    // 账户内金额
    private int amount;

    // 增加账户金额
    public void addAmount(int amount) {this.amount = this.amount + amount;}
    // 省略构造方法和 getter/setter 
}

创立一个线程实现类 AccountOperationThread:

public class AccountOperationThread implements Runnable {private final static Logger logger = LoggerFactory.getLogger(AccountOperationThread.class);

    private static final Long RELEASE_SUCCESS = 1L;

    private String userId;

    private RedisTemplate<Object, Object> redisTemplate;

    public AccountOperationThread(String userId, RedisTemplate<Object, Object> redisTemplate) {
        this.userId = userId;
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void run() {noLock();
    }

    /**
     * 不加锁
     */
    private void noLock() {
        try {Random random = new Random();
            // 模仿线程进行业务解决
            TimeUnit.MILLISECONDS.sleep(random.nextInt(100) + 1);
        } catch (InterruptedException e) {e.printStackTrace();
        }
        // 模仿数据库中获取用户账号
        UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId);
        // 金额 +1
        userAccount.addAmount(1);
        logger.info(Thread.currentThread().getName() + ": user id :" + userId + "amount :" + userAccount.getAmount());
        // 模仿存回数据库
        redisTemplate.opsForValue().set(userId, userAccount);
    }
}

其中 RedisTemplate 的实例化交给了 Spring Boot:

@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer =
                new Jackson2JsonRedisSerializer<>(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        // 设置 value 的序列化规定和 key 的序列化规定
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }
}

最初,再筹备一个 TestController 来进行触发多线程的运行:

@RestController
public class TestController {private final static Logger logger = LoggerFactory.getLogger(TestController.class);

    private static ExecutorService executorService = Executors.newFixedThreadPool(10);

    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;

    @GetMapping("/test")
    public String test() throws InterruptedException {
        // 初始化用户 user_001 到 Redis,账户金额为 0
        redisTemplate.opsForValue().set("user_001", new UserAccount("user_001", 0));
        // 开启 10 个线程进行同步测试,每个线程为账户减少 1 元
        for (int i = 0; i < 10; i++) {logger.info("创立线程 i =" + i);
            executorService.execute(new AccountOperationThread("user_001", redisTemplate));
        }

        // 主线程休眠 1 秒期待线程跑完
        TimeUnit.MILLISECONDS.sleep(1000);
        // 查问 Redis 中的 user_001 账户
        UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get("user_001");
        logger.info("user id :" + userAccount.getUserId() + "amount :" + userAccount.getAmount());
        return "success";
    }
}

执行上述程序,失常来说 10 个线程,每个线程加 1,后果应该是 10。但多执行几次,会发现,后果变化很大,基本上都要比 10 小。

[pool-1-thread-5] c.s.redis.thread.AccountOperationThread  : pool-1-thread-5 : user id : user_001 amount : 1
[pool-1-thread-4] c.s.redis.thread.AccountOperationThread  : pool-1-thread-4 : user id : user_001 amount : 1
[pool-1-thread-3] c.s.redis.thread.AccountOperationThread  : pool-1-thread-3 : user id : user_001 amount : 1
[pool-1-thread-1] c.s.redis.thread.AccountOperationThread  : pool-1-thread-1 : user id : user_001 amount : 1
[pool-1-thread-1] c.s.redis.thread.AccountOperationThread  : pool-1-thread-1 : user id : user_001 amount : 2
[pool-1-thread-2] c.s.redis.thread.AccountOperationThread  : pool-1-thread-2 : user id : user_001 amount : 2
[pool-1-thread-5] c.s.redis.thread.AccountOperationThread  : pool-1-thread-5 : user id : user_001 amount : 2
[pool-1-thread-4] c.s.redis.thread.AccountOperationThread  : pool-1-thread-4 : user id : user_001 amount : 3
[pool-1-thread-1] c.s.redis.thread.AccountOperationThread  : pool-1-thread-1 : user id : user_001 amount : 4
[pool-1-thread-3] c.s.redis.thread.AccountOperationThread  : pool-1-thread-3 : user id : user_001 amount : 5
[nio-8080-exec-1] c.s.redis.controller.TestController      : user id : user_001 amount : 5

以上述日志为例,前四个线程都将值改为 1,也就是前面三个线程都将后面的批改进行了笼罩,导致最终后果不是 10,只有 5。这显然是有问题的。

Redis 同步锁实现

针对下面的状况,在同一个 JVM 当中,咱们能够通过线程加锁来实现。但在分布式环境下,JVM 级别的锁是没方法实现的,这里能够采纳 Redis 同步锁实现。

基本思路:第一个线程进入时,在 Redis 中进记录,当后续线程过去申请时,判断 Redis 是否存在该记录,如果存在则阐明处于锁定状态,进行期待或返回。如果不存在,则进行后续业务解决。

  /**
     * 1. 抢占资源时判断是否被锁。* 2. 如未锁则抢占胜利且加锁,否则期待锁开释。* 3. 业务实现后开释锁, 让给其它线程。* <p>
     * 该计划并未解决同步问题,起因:线程取得锁和加锁的过程,并非原子性操作,可能会导致线程 A 取得锁,还未加锁时,线程 B 也取得了锁。*/
    private void redisLock() {Random random = new Random();
        try {TimeUnit.MILLISECONDS.sleep(random.nextInt(1000) + 1);
        } catch (InterruptedException e) {e.printStackTrace();
        }
        while (true) {Object lock = redisTemplate.opsForValue().get(userId + ":syn");
            if (lock == null) {
                // 取得锁 -> 加锁 -> 跳出循环
                logger.info(Thread.currentThread().getName() + ": 取得锁");
                redisTemplate.opsForValue().set(userId + ":syn", "lock");
                break;
            }
            try {
                // 期待 500 毫秒重试取得锁
                TimeUnit.MILLISECONDS.sleep(500);
            } catch (InterruptedException e) {e.printStackTrace();
            }
        }
        try {
            // 模仿数据库中获取用户账号
            UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId);
            if (userAccount != null) {
                // 设置金额
                userAccount.addAmount(1);
                logger.info(Thread.currentThread().getName() + ": user id :" + userId + "amount :" + userAccount.getAmount());
                // 模仿存回数据库
                redisTemplate.opsForValue().set(userId, userAccount);
            }
        } finally {
            // 开释锁
            redisTemplate.delete(userId + ":syn");
            logger.info(Thread.currentThread().getName() + ": 开释锁");
        }
    }

在 while 代码块中,先判断对应用户 ID 是否在 Redis 中存在,如果不存在,则进行 set 加锁,如果存在,则跳出循环持续期待。

上述代码,看起来实现了加锁的性能,但当执行程序时,会发现与未加锁一样,仍旧存在并发问题。起因是:获取锁和加锁的操作并不是原子的。比方两个线程发现 lock 都是 null,都进行了加锁,此时并发问题仍旧存在。

Redis 原子性同步锁

针对上述问题,可将获取锁和加锁的过程原子化解决。基于 spring-boot-data-redis 提供的原子化 API 能够实现:

// 该办法应用了 redis 的指令:SETNX key value
// 1.key 不存在,设置胜利返回 value,setIfAbsent 返回 true;// 2.key 存在,则设置失败返回 null,setIfAbsent 返回 false;// 3. 原子性操作;Boolean setIfAbsent(K var1, V var2);

上述办法的原子化操作是对 Redis 的 setnx 命令的封装,在 Redis 中 setnx 的应用如下实例:

redis> SETNX mykey "Hello"
(integer) 1
redis> SETNX mykey "World"
(integer) 0
redis> GET mykey
"Hello"

第一次,设置 mykey 时,并不存在,则返回 1,示意设置胜利;第二次设置 mykey 时,曾经存在,则返回 0,示意设置失败。再次查问 mykey 对应的值,会发现仍旧是第一次设置的值。也就是说 redis 的 setnx 保障了惟一的 key 只能被一个服务设置胜利。

了解了上述 API 及底层原理,来看看线程中的实现办法代码如下:

    /**
     * 1. 原子操作加锁
     * 2. 竞争线程循环重试取得锁
     * 3. 业务实现开释锁
     */
    private void atomicityRedisLock() {
        //Spring data redis 反对的原子性操作
        while (!redisTemplate.opsForValue().setIfAbsent(userId + ":syn", "lock")) {
            try {
                // 期待 100 毫秒重试取得锁
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {e.printStackTrace();
            }
        }
        logger.info(Thread.currentThread().getName() + ": 取得锁");
        try {
            // 模仿数据库中获取用户账号
            UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId);
            if (userAccount != null) {
                // 设置金额
                userAccount.addAmount(1);
                logger.info(Thread.currentThread().getName() + ": user id :" + userId + "amount :" + userAccount.getAmount());
                // 模仿存回数据库
                redisTemplate.opsForValue().set(userId, userAccount);
            }
        } finally {
            // 开释锁
            redisTemplate.delete(userId + ":syn");
            logger.info(Thread.currentThread().getName() + ": 开释锁");
        }
    }

再次执行代码,会发现后果正确了,也就是说能够胜利的对分布式线程进行了加锁。

Redis 分布式锁的死锁

尽管上述代码执行后果没问题,但如果利用异样宕机,没来得及执行 finally 中开释锁的办法,那么其余线程则永远无奈取得这个锁。

此时可采纳 setIfAbsent 的重载办法:

Boolean setIfAbsent(K var1, V var2, long var3, TimeUnit var5);

基于该办法,能够设置锁的过期工夫。这样即使取得锁的线程宕机,在 Redis 中数据过期之后,其余线程可失常取得该锁。

示例代码如下:

private void atomicityAndExRedisLock() {
        try {
            //Spring data redis 反对的原子性操作, 并设置 5 秒过期工夫
            while (!redisTemplate.opsForValue().setIfAbsent(userId + ":syn",
                    System.currentTimeMillis() + 5000, 5000, TimeUnit.MILLISECONDS)) {
                // 期待 100 毫秒重试取得锁
                logger.info(Thread.currentThread().getName() + ": 尝试循环获取锁");
                TimeUnit.MILLISECONDS.sleep(1000);
            }
            logger.info(Thread.currentThread().getName() + ": 取得锁 --------");
            // 利用在这里宕机,过程退出,无奈执行 finally;
            Thread.currentThread().interrupt();
            // 业务逻辑...
        } catch (InterruptedException e) {e.printStackTrace();
        } finally {
            // 开释锁
            if (!Thread.currentThread().isInterrupted()) {redisTemplate.delete(userId + ":syn");
                logger.info(Thread.currentThread().getName() + ": 开释锁");
            }
        }
    }

业务超时及守护线程

下面增加了 Redis 所的超时工夫,看似解决了问题,但又引入了新的问题。

比方,失常状况下线程 A 在 5 秒内可失常解决完业务,但偶发会呈现超过 5 秒的状况。如果将超时工夫设置为 5 秒,线程 A 取得了锁,但业务逻辑解决须要 6 秒。此时,线程 A 还在失常业务逻辑,线程 B 曾经取得了锁。当线程 A 解决完时,有可能将线程 B 的锁给开释掉。

在上述场景中有两个问题点:

  • 第一,线程 A 和线程 B 可能会同时在执行,存在并发问题。
  • 第二,线程 A 可能会把线程 B 的锁给开释掉,导致一系列的恶性循环。

当然,能够通过在 Redis 中设置 value 值来判断锁是属于线程 A 还是线程 B。但仔细分析会发现,这个问题的实质是因为线程 A 执行业务逻辑耗时超出了锁超时的工夫。

那么就有两个解决方案了:

  • 第一,将超时工夫设置的足够长,确保业务代码可能在锁开释之前执行实现;
  • 第二,为锁增加守护线程,为将要过期开释但未开释的锁减少工夫;

第一种形式须要全行大多数状况下业务逻辑的耗时,进行超时工夫的设定。

第二种形式,可通过如下守护线程的形式来动静减少锁超时工夫。

public class DaemonThread implements Runnable {private final static Logger logger = LoggerFactory.getLogger(DaemonThread.class);

    // 是否须要守护 主线程敞开则完结守护线程
    private volatile boolean daemon = true;
    // 守护锁
    private String lockKey;

    private RedisTemplate<Object, Object> redisTemplate;

    public DaemonThread(String lockKey, RedisTemplate<Object, Object> redisTemplate) {
        this.lockKey = lockKey;
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void run() {
        try {while (daemon) {long time = redisTemplate.getExpire(lockKey, TimeUnit.MILLISECONDS);
                // 残余有效期小于 1 秒则续命
                if (time < 1000) {logger.info("守护过程:" + Thread.currentThread().getName() + "缩短锁工夫 5000 毫秒");
                    redisTemplate.expire(lockKey, 5000, TimeUnit.MILLISECONDS);
                }
                TimeUnit.MILLISECONDS.sleep(300);
            }
            logger.info("守护过程:" + Thread.currentThread().getName() + "敞开");
        } catch (InterruptedException e) {e.printStackTrace();
        }
    }

    // 主线程被动调用完结
    public void stop() {daemon = false;}
}

上述线程每隔 300 毫秒获取一下 Redis 中锁的超时工夫,如果小于 1 秒,则缩短 5 秒。当主线程调用敞开时,守护线程也随之敞开。

主线程中相干代码实现:

private void deamonRedisLock() {
        // 守护线程
        DaemonThread daemonThread = null;
        //Spring data redis 反对的原子性操作, 并设置 5 秒过期工夫
        String uuid = UUID.randomUUID().toString();
        String value = Thread.currentThread().getId() + ":" + uuid;
        try {while (!redisTemplate.opsForValue().setIfAbsent(userId + ":syn", value, 5000, TimeUnit.MILLISECONDS)) {
                // 期待 100 毫秒重试取得锁
                logger.info(Thread.currentThread().getName() + ": 尝试循环获取锁");
                TimeUnit.MILLISECONDS.sleep(1000);
            }
            logger.info(Thread.currentThread().getName() + ": 取得锁 ----");
            // 开启守护线程
            daemonThread = new DaemonThread(userId + ":syn", redisTemplate);
            Thread thread = new Thread(daemonThread);
            thread.start();
            // 业务逻辑执行 10 秒...
            TimeUnit.MILLISECONDS.sleep(10000);
        } catch (InterruptedException e) {e.printStackTrace();
        } finally {
            // 开释锁 这里也须要原子操作, 今后通过 Redis + Lua 讲
            String result = (String) redisTemplate.opsForValue().get(userId + ":syn");
            if (value.equals(result)) {redisTemplate.delete(userId + ":syn");
                logger.info(Thread.currentThread().getName() + ": 开释锁 -----");
            }
            // 敞开守护线程
            if (daemonThread != null) {daemonThread.stop();
            }
        }
    }

其中在取得锁之后,开启守护线程,在 finally 中将守护线程敞开。

基于 Lua 脚本的实现

在上述逻辑中,咱们是基于 spring-boot-data-redis 提供的原子化操作来保障锁判断和执行的原子化的。在非 Spring Boot 我的项目中,则能够基于 Lua 脚本来实现。

首先定义加锁和解锁的 Lua 脚本及对应的 DefaultRedisScript 对象,在 RedisConfig 配置类中增加如下实例化代码:

@Configuration
public class RedisConfig {

    //lock script
    private static final String LOCK_SCRIPT = "if redis.call('setnx',KEYS[1],ARGV[1]) == 1" +
            "then redis.call('expire',KEYS[1],ARGV[2])" +
            "return 1" +
            "else return 0 end";
    private static final String UNLOCK_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call" +
            "('del', KEYS[1]) else return 0 end";

    // ... 省略局部代码
    
    @Bean
    public DefaultRedisScript<Boolean> lockRedisScript() {DefaultRedisScript<Boolean> defaultRedisScript = new DefaultRedisScript<>();
        defaultRedisScript.setResultType(Boolean.class);
        defaultRedisScript.setScriptText(LOCK_SCRIPT);
        return defaultRedisScript;
    }

    @Bean
    public DefaultRedisScript<Long> unlockRedisScript() {DefaultRedisScript<Long> defaultRedisScript = new DefaultRedisScript<>();
        defaultRedisScript.setResultType(Long.class);
        defaultRedisScript.setScriptText(UNLOCK_SCRIPT);
        return defaultRedisScript;
    }
}

再通过在 AccountOperationThread 类中新建构造方法,将上述两个对象传入类中(省略此局部演示)。而后,就能够基于 RedisTemplate 来调用了,革新之后的代码实现如下:

    private void deamonRedisLockWithLua() {
        // 守护线程
        DaemonThread daemonThread = null;
        //Spring data redis 反对的原子性操作, 并设置 5 秒过期工夫
        String uuid = UUID.randomUUID().toString();
        String value = Thread.currentThread().getId() + ":" + uuid;
        try {while (!redisTemplate.execute(lockRedisScript, Collections.singletonList(userId + ":syn"), value, 5)) {
                // 期待 1000 毫秒重试取得锁
                logger.info(Thread.currentThread().getName() + ": 尝试循环获取锁");
                TimeUnit.MILLISECONDS.sleep(1000);
            }
            logger.info(Thread.currentThread().getName() + ": 取得锁 ----");
            // 开启守护线程
            daemonThread = new DaemonThread(userId + ":syn", redisTemplate);
            Thread thread = new Thread(daemonThread);
            thread.start();
            // 业务逻辑执行 10 秒...
            TimeUnit.MILLISECONDS.sleep(10000);
        } catch (InterruptedException e) {logger.error("异样", e);
        } finally {
            // 应用 Lua 脚本:先判断是否是本人设置的锁,再执行删除
            // key 存在, 以后值 = 期望值时, 删除 key;key 存在, 以后值!= 期望值时, 返回 0;
            Long result = redisTemplate.execute(unlockRedisScript, Collections.singletonList(userId + ":syn"), value);
            logger.info("redis 解锁:{}", RELEASE_SUCCESS.equals(result));
            if (RELEASE_SUCCESS.equals(result)) {if (daemonThread != null) {
                    // 敞开守护线程
                    daemonThread.stop();
                    logger.info(Thread.currentThread().getName() + ": 开释锁 ---");
                }
            }
        }
    }

其中 while 循环中加锁和 finally 中的开释锁都是基于 Lua 脚本来实现了。

Redis 锁的其余因素

除了上述实例,在应用 Redis 分布式锁时,还能够思考以下状况及计划。

Redis 锁的不可重入

当线程在持有锁的状况下再次申请加锁,如果一个锁反对一个线程屡次加锁,那么这个锁就是可重入的。如果一个不可重入锁被再次加锁,因为该锁曾经被持有,再次加锁会失败。Redis 可通过对锁进行重入计数,加锁时加 1,解锁时减 1,当计数归 0 时开释锁。

可重入锁尽管高效但会减少代码的复杂性,这里就不举例说明了。

期待锁开释

有的业务场景,发现被锁则间接返回。但有的场景下,客户端须要期待锁开释而后去抢锁。上述示例就属于后者。针对期待锁开释也有两种计划:

  • 客户端轮训:当未取得锁时,期待一段时间再从新获取,直到胜利。上述示例就是基于这种形式实现的。这种形式的毛病也很显著,比拟消耗服务器资源,当并发量大时会影响服务器的效率。
  • 应用 Redis 的订阅公布性能:当获取锁失败时,订阅锁开释音讯,获取锁胜利后开释时,发送开释音讯。

集群中的主备切换和脑裂

在 Redis 蕴含主从同步的集群部署形式中,如果主节点挂掉,从节点晋升为主节点。如果客户端 A 在主节点加锁胜利,指令还未同步到从节点,此时主节点挂掉,从节点升为主节点,新的主节点中没有锁的数据。这种状况下,客户端 B 就可能加锁胜利,从而呈现并发的场景。

当集群产生脑裂时,Redis master 节点跟 slave 节点和 sentinel 集群处于不同的网络分区。sentinel 集群无奈感知到 master 的存在,会将 slave 节点晋升为 master 节点,此时就会存在两个不同的 master 节点。从而也会导致并发问题的呈现。Redis Cluster 集群部署形式同理。

小结

通过生产环境中的一个问题,排查起因,寻找解决方案,到最终对基于 Redis 分布式的深入研究,这便是学习的过程。

同时,每当面试或被问题如何解决分布式共享资源时,咱们会脱口而出”基于 Redis 实现分布式锁“,但通过本文的学习会发现,Redis 分布式锁并不是万能的,而且在应用的过程中还须要留神超时、死锁、误会锁、集群选主 / 脑裂等问题。

Redis 以高性能著称,但在实现分布式锁的过程中还是存在一些问题。因而,基于 Redis 的分布式锁能够极大的缓解并发问题,但要齐全避免并发,还是得从数据库层面动手。

源码地址:https://github.com/secbr/spri…

博主简介:《SpringBoot 技术底细》技术图书作者,热爱钻研技术,写技术干货文章。

公众号:「程序新视界」,博主的公众号,欢送关注~

技术交换:请分割博主微信号:zhuan2quan

正文完
 0