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

既然我的项目交付到手中，这样的问题是必须要解决的。梳理了所有账务解决逻辑，最终找到了起因：数据库并发操作热点账户导致。就这这个问题，来聊一聊分布式系统下基于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：

@Configurationpublic 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来进行触发多线程的运行：

@RestControllerpublic 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) 1redis> SETNX mykey "World"(integer) 0redis> 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配置类中增加如下实例化代码：

@Configurationpublic 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