老大看了都摇头系列：使用Redis实现一把分布式锁

结构化程序员 2022-04-06

261

1. 分布式锁

1.1 概述

分布式应用使用的锁。

1.1.1 应用场景

互联网秒杀
抢优惠券
接口幂等性校验

1.2 使用Redis实现

version-1

业务代码出现异常，锁无法释放

public String deductStock() {
    String lockKey = "product:001";
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, lockKey);
    if (!result) {
        return "error";
    }
    
    int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
    if (stock > 0) {
        // 可能是复杂的业务处理代码
        int realStock = stock - 1;
        stringRedisTemplate.opsForValue().set("stock", realStock + "");
        System.out.println("扣减成功，剩余库存" + realStock);
    } else {
        System.out.println("扣减失败，库存不足");
    }
    
    stringRedisTemplate.delete(lockKey);
    
    return "end";
}

version-2

业务代码执行到一半，JVM宕机了，锁无法释放

public String deductStock() {
        String lockKey = "product:001";
        try {
            Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, lockKey, 30, TimeUnit.SECONDS);
            if (!result) {
                return "error";
            }


            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if (stock > 0) {
                // 可能是复杂的业务处理代码
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + "");
                System.out.println("扣减成功，剩余库存" + realStock);
            } else {
                System.out.println("扣减失败，库存不足");
            }
        } finally {
            stringRedisTemplate.delete(lockKey);
        }


        return "end";
    }

version-3

假设超时时间设置为30秒，业务代码执行完需要35秒，在业务代码执行期间，Redis把锁释放了，那么其他线程又能继续访问临界资源了，高并发场景下，会存在锁失效问题。
另外，由于当前线程释放了锁，假如其他线程执行这段代码，只需要3秒，那么他就可以在3秒后把锁删除掉，即其他线程删除了当前线程的锁。

public String deductStock() {
        String lockKey = "product:001";
        try {
            Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, lockKey, 30, TimeUnit.SECONDS);
            if (!result) {
                return "error";
            }


            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if (stock > 0) {
                // 可能是复杂的业务处理代码
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + "");
                System.out.println("扣减成功，剩余库存" + realStock);
            } else {
                System.out.println("扣减失败，库存不足");
            }
        } finally {
            stringRedisTemplate.delete(lockKey);
        }


        return "end";
    }

version-final

如果集群架构下，master节点没有同步锁到slaver，那么即使选举出最新的slaver，锁仍然失效了。

public String deductStock() {
        String lockKey = "product:001";
        String clientId = UUID.randomUUID().toString();
        // try-catch 防止异常无法释放锁
        try {
            // 超时时间防止宕机无法释放分布式锁
            Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);
            if (!result) {
                return "error";
            }


            // 续命，防止业务执行过久，提前释放分布式锁
            new Timer().scheduleAtFixedRate(new TimerTask() {
                @Override
                public void run() {
                    String result = stringRedisTemplate.opsForValue().get(lockKey);
                    if (result != null) {
                        stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);
                    }
                }
            }, 10, 10);


            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if (stock > 0) {
                // 可能是复杂的业务处理代码
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + "");
                System.out.println("扣减成功，剩余库存" + realStock);
            } else {
                System.out.println("扣减失败，库存不足");
            }
        } finally {
            // 防止其他线程删除当前线程分布式锁
            String value = stringRedisTemplate.opsForValue().get(lockKey);
            if (value.equals(clientId))
                stringRedisTemplate.delete(lockKey);
        }


        return "end";
    }

1.3 总结

实现一把完善的Redis锁需要考虑各种异常情况，比如代码异常、系统宕机、锁过期、集群master重新选举。
Redis作为简单架构下的分布式锁实现已经足够，但是集群架构下，Redis设计初衷是为了实现缓存功能，如果硬是为了实现CP架构实现分布式锁，本身会与AP架构的高性能冲突，另外代码为了完善分布式锁的各种异常处理，还会增加复杂性，复杂性再次带来了性能的损耗。
一个系统的设计是要结合：组件架构、代码异常处理、性能、复杂性各个因素进行权衡，最终完成一个适合业务的软件。

2. Redisson

2.1 概述

Redisson是架设在Redis基础上的一个Java驻内存数据网格（In-Memory Data Grid）。【Redis官方推荐】

Redisson在基于NIO的Netty框架上，充分的利用了Redis键值数据库提供的一系列优势，在Java实用工具包中常用接口的基础上，为使用者提供了一系列具有分布式特性的常用工具类。使得原本作为协调单机多线程并发程序的工具包获得了协调分布式多机多线程并发系统的能力，大大降低了设计和研发大规模分布式系统的难度。同时结合各富特色的分布式服务，更进一步简化了分布式环境中程序相互之间的协作。

2.2 使用

使用Redisson实现一把分布式锁。

@Autowired
private Redisson redisson;


@Bean
public Redisson redisson() {
    // 单机模式
    Config config = new Config();
    config.useSingleServer().setAddress("redis://localhost:6379").setDatabase(0);
    return ((Redisson) Redisson.create(config));
}


public String deductStock2() {
    String lockKey = "product:001";
    RLock lock = redisson.getLock(lockKey);


    try {
        // 默认设置了超时时间
        lock.lock();  //setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if (stock > 0) {
            // 可能是复杂的业务处理代码
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock", realStock + "");
            System.out.println("扣减成功，剩余库存" + realStock);
        } else {
            System.out.println("扣减失败，库存不足");
        }
    } finally {
        lock.unlock();
        /*if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {
            stringRedisTemplate.delete(lockKey);
        }*/
    }


    return "end";
}

2.3 原理

2.4 RedLock

实现CP架构的分布式锁，不推荐使用。

public String redlock() {
    String lockKey = "product:001";
    //这里需要自己实例化不同redis实例的redisson客户端连接，这里只是伪代码用一个redisson客户端简化了
    RLock lock1 = redisson.getLock(lockKey);
    RLock lock2 = redisson.getLock(lockKey);
    RLock lock3 = redisson.getLock(lockKey);


    /**
     * 根据多个 RLock 对象构建 RedissonRedLock （最核心的差别就在这里）
     */
    RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
    try {
        /**
         * waitTimeout 尝试获取锁的最大等待时间，超过这个值，则认为获取锁失败
         * leaseTime   锁的持有时间,超过这个时间锁会自动失效（值应设置为大于业务处理的时间，确保在锁有效期内业务能处理完）
         */
        boolean res = redLock.tryLock(10, 30, TimeUnit.SECONDS);
        if (res) {
            //成功获得锁，在这里处理业务
        }
    } catch (Exception e) {
        throw new RuntimeException("lock fail");
    } finally {
        //无论如何, 最后都要解锁
        redLock.unlock();
    }


    return "end";
}

redis

文章转载自结构化程序员，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。