分布式锁实现:Redis

作者: Java黎先生 | 来源:发表于2018-08-22 17:25 被阅读3次

    前言

    单机环境下我们可以通过JAVA的Synchronized和Lock来实现进程内部的锁,但是随着分布式应用和集群环境的出现,系统资源的竞争从单进程多线程的竞争变成了多进程的竞争,这时候就需要分布式锁来保证。

    实现分布式锁现在主流的方式大致有以下三种

    1. 基于数据库的索引和行锁

    2. 基于Redis的单线程原子操作:setNX

    3. 基于Zookeeper的临时有序节点

    这篇文章我们用Redis来实现,会基于现有的各种锁实现来分析,最后分享Redission的锁源码分析来看下分布式锁的开源实现

    复制代码

    设计实现

    加锁

    一、 通过setNx和getSet来实现

    这是现在网上大部分版本的实现方式,笔者之前项目里面用到分布式锁也是通过这样的方式实现

    public boolean lock(Jedis jedis, String lockName, Integer expire) {

    //返回是否设置成功

    //setNx加锁

    long now = System.currentTimeMillis();

    boolean result = jedis.setnx(lockName, String.valueOf(now + expire * 1000)) == 1;

    if (!result) {

    //防止死锁的容错

    String timestamp = jedis.get(lockName);

    if (timestamp != null && Long.parseLong(timestamp) < now) {

    //不通过del方法来删除锁。而是通过同步的getSet

    String oldValue = jedis.getSet(lockName, String.valueOf(now + expire));

    if (oldValue != null && oldValue.equals(timestamp)) {

    result = true;

    jedis.expire(lockName, expire);

    }

    }

    }

    if (result) {

    jedis.expire(lockName, expire);

    }

    return result;

    }

    代码分析

    通过setNx命令老保证操作的原子性,获取到锁,并且把过期时间设置到value里面

    通过expire方法设置过期时间,如果设置过期时间失败的话,再通过value的时间戳来和当前时间戳比较,防止出现死锁

    通过getSet命令在发现锁过期未被释放的情况下,避免删除了在这个过程中有可能被其余的线程获取到了锁

    存在问题

    防止死锁的解决方案是通过系统当前时间决定的,不过线上服务器系统时间一般来说都是一致的,这个不算是严重的问题

    锁过期的时候可能会有多个线程执行getSet命令,在竞争的情况下,会修改value的时间戳,理论上来说会有误差

    锁无法具备客户端标识,在解锁的时候可能被其余的客户端删除同一个key

    虽然有小问题,不过大体上来说这种分布式锁的实现方案基本上是符合要求的,能够做到锁的互斥和避免死锁

    二、 通过Redis高版本的原子命令

    jedis的set命令可以自带复杂参数,通过这些参数可以实现原子的分布式锁命令

    jedis.set(lockName, "", "NX", "PX", expireTime);

    复制代码

    代码分析

    redis的set命令可以携带复杂参数,第一个是锁的key,第二个是value,可以存放获取锁的客户端ID,通过这个校验是否当前客户端获取到了锁,第三个参数取值NX/XX,第四个参数 EX|PX,第五个就是时间

    NX:如果不存在就设置这个key XX:如果存在就设置这个key

    EX:单位为秒,PX:单位为毫秒

    这个命令实质上就是把我们之前的setNx和expire命令合并成一个原子操作命令,不需要我们考虑set失败或者expire失败的情况

    解锁

    一、 通过Redis的del命令

    public boolean unlock(Jedis jedis, String lockName) {

    jedis.del(lockName);

    return true;

    }

    代码分析

    通过redis的del命令可以直接删除锁,可能会出现误删其他线程已经存在的锁的情况

    二、 Redis的del检查

    public static void unlock2(Jedis jedis, String lockKey, String requestId) {

    // 判断加锁与解锁是不是同一个客户端

    if (requestId.equals(jedis.get(lockKey))) {

    // 若在此时,这把锁突然不是这个客户端的,则会误解锁

    jedis.del(lockKey);

    }

    }

    代码分析

    新增了requestId客户端ID的判断,但由于不是原子操作,在多个进程下面的并发竞争情况下,无法保证安全

    三、 Redis的LUA脚本

    public static boolean unlock3(Jedis jedis, String lockKey, String requestId) {

    String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

    Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(""));

    if (1L == (long) result) {

    return true;

    }

    return false;

    }

    代码分析

    通过Lua脚本来保证操作的原子性,其实就是把之前的先判断再删除合并成一个原子性的脚本命令,逻辑就是,先通过get判断value是不是相等,若相等就删除,否则就直接return

    Redission的分布式锁

    Redission是redis官网推荐的一个redis客户端,除了基于redis的基础的CURD命令以外,重要的是就是Redission提供了方便好用的分布式锁API

    复制代码

    一、 基本用法

    RedissonClient redissonClient = RedissonTool.getInstance();

    RLock distribute_lock = redissonClient.getLock("distribute_lock");

    try {

    boolean result = distribute_lock.tryLock(3, 10, TimeUnit.SECONDS);

    } catch (InterruptedException e) {

    e.printStackTrace();

    } finally {

    if (distribute_lock.isLocked()) {

    distribute_lock.unlock();

    }

    }

    代码流程

    通过redissonClient获取RLock实例

    tryLock获取尝试获取锁,第一个是等待时间,第二个是锁的超时时间,第三个是时间单位

    执行完业务逻辑后,最终释放锁

    二、 具体实现

    我们通过tryLock来分析redission分布式的实现,lock方法跟tryLock差不多,只不过没有最长等待时间的设置,会自旋循环等待锁的释放,直到获取锁为止

    long time = unit.toMillis(waitTime);

    long current = System.currentTimeMillis();

    //获取当前线程ID,用于实现可重入锁

    final long threadId = Thread.currentThread().getId();

    //尝试获取锁

    Long ttl = tryAcquire(leaseTime, unit, threadId);

    // lock acquired

    if (ttl == null) {

    return true;

    }

    time -= (System.currentTimeMillis() - current);

    if (time <= 0) {

    //等待时间结束,返回获取失败

    acquireFailed(threadId);

    return false;

    }

    current = System.currentTimeMillis();

    //订阅锁的队列,等待锁被其余线程释放后通知

    final RFuture subscribeFuture = subscribe(threadId);

    if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {

    if (!subscribeFuture.cancel(false)) {

    subscribeFuture.addListener(new FutureListener() {

    @Override

    public void operationComplete(Future future) throws Exception {

    if (subscribeFuture.isSuccess()) {

    unsubscribe(subscribeFuture, threadId);

    }

    }

    });

    }

    acquireFailed(threadId);

    return false;

    }

    try {

    time -= (System.currentTimeMillis() - current);

    if (time <= 0) {

    acquireFailed(threadId);

    return false;

    }

    while (true) {

    long currentTime = System.currentTimeMillis();

    ttl = tryAcquire(leaseTime, unit, threadId);

    // lock acquired

    if (ttl == null) {

    return true;

    }

    time -= (System.currentTimeMillis() - currentTime);

    if (time <= 0) {

    acquireFailed(threadId);

    return false;

    }

    // waiting for message,等待订阅的队列消息

    currentTime = System.currentTimeMillis();

    if (ttl >= 0 && ttl < time) {

    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);

    } else {

    getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);

    }

    time -= (System.currentTimeMillis() - currentTime);

    if (time <= 0) {

    acquireFailed(threadId);

    return false;

    }

    }

    } finally {

    unsubscribe(subscribeFuture, threadId);

    }

    代码分析

    首先tryAcquire尝试获取锁,若返回ttl为null,说明获取到锁了

    判断等待时间是否过期,如果过期,直接返回获取锁失败

    通过Redis的Channel订阅监听队列,subscribe内部通过信号量semaphore,再通过await方法阻塞,内部其实是用CountDownLatch来实现阻塞,获取subscribe异步执行的结果,来保证订阅成功,再判断是否到了等待时间

    再次尝试申请锁和等待时间的判断,循环阻塞在这里等待锁释放的消息RedissonLockEntry也维护了一个semaphore的信号量

    无论是否释放锁,最终都要取消订阅这个队列消息

    redission内部的getEntryName是客户端实例ID+锁名称来保证多个实例下的锁可重入

    tryAcquire获取锁

    redisssion获取锁的核心代码,内部其实是异步调用,但是用get方法阻塞了

    private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {

    return get(tryAcquireAsync(leaseTime, unit, threadId));

    }

    private RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {

    if (leaseTime != -1) {

    return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);

    }

    RFuture ttlRemainingFuture = tryLockInnerAsync(LOCK_EXPIRATION_INTERVAL_SECONDS, TimeUnit.SECONDS, threadId, RedisCommands.EVAL_LONG);

    ttlRemainingFuture.addListener(new FutureListener() {

    @Override

    public void operationComplete(Future future) throws Exception {

    if (!future.isSuccess()) {

    return;

    }

    Long ttlRemaining = future.getNow();

    // lock acquired

    if (ttlRemaining == null) {

    scheduleExpirationRenewal(threadId);

    }

    }

    });

    return ttlRemainingFuture;

    }

    tryLockInnerAsync方法内部是基于Lua脚本来获取锁的

    先判断KEYS[1](锁名称)对应的key是否存在,不存在获取到锁,hset设置key的value,pexpire设置过期时间,返回null表示获取到锁

    存在的话,锁被占,hexists判断是否是当前线程的锁,若是的话,hincrby增加重入次数,重新设置过期时间,不是当前线程的锁,返回当前锁的过期时间

    RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {

    internalLockLeaseTime = unit.toMillis(leaseTime);

    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,

    "if (redis.call('exists', KEYS[1]) == 0) then " +

    "redis.call('hset', KEYS[1], ARGV[2], 1); " +

    "redis.call('pexpire', KEYS[1], ARGV[1]); " +

    "return nil; " +

    "end; " +

    "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +

    "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +

    "redis.call('pexpire', KEYS[1], ARGV[1]); " +

    "return nil; " +

    "end; " +

    "return redis.call('pttl', KEYS[1]);",

    Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));

    }

    复制代码

    Redission避免死锁的解决方案:

    Redission为了避免锁未被释放,采用了一个特殊的解决方案,若未设置过期时间的话,redission默认的过期时间是30s,同时未避免锁在业务未处理完成之前被提前释放,Redisson在获取到锁且默认过期时间的时候,会在当前客户端内部启动一个定时任务,每隔internalLockLeaseTime/3的时间去刷新key的过期时间,这样既避免了锁提前释放,同时如果客户端宕机的话,这个锁最多存活30s的时间就会自动释放(刷新过期时间的定时任务进程也宕机)

    // lock acquired,获取到锁的时候设置定期更新时间的任务

    if (ttlRemaining) {

    scheduleExpirationRenewal(threadId);

    }

    //expirationRenewalMap的并发安全MAP记录设置过的缓存,避免并发情况下重复设置任务,internalLockLeaseTime / 3的时间后重新设置过期时间

    private void scheduleExpirationRenewal(final long threadId) {

    if (expirationRenewalMap.containsKey(getEntryName())) {

    return;

    }

    Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {

    @Override

    public void run(Timeout timeout) throws Exception {

    RFuture future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,

    "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +

    "redis.call('pexpire', KEYS[1], ARGV[1]); " +

    "return 1; " +

    "end; " +

    "return 0;",

    Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));

    future.addListener(new FutureListener() {

    @Override

    public void operationComplete(Future future) throws Exception {

    expirationRenewalMap.remove(getEntryName());

    if (!future.isSuccess()) {

    log.error("Can't update lock " + getName() + " expiration", future.cause());

    return;

    }

    if (future.getNow()) {

    // reschedule itself

    scheduleExpirationRenewal(threadId);

    }

    }

    });

    }

    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

    if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {

    task.cancel();

    }

    }

    unlock解锁

    protected RFuture unlockInnerAsync(long threadId) {

    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,

    "if (redis.call('exists', KEYS[1]) == 0) then " +

    "redis.call('publish', KEYS[2], ARGV[1]); " +

    "return 1; " +

    "end;" +

    "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +

    "return nil;" +

    "end; " +

    "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +

    "if (counter > 0) then " +

    "redis.call('pexpire', KEYS[1], ARGV[2]); " +

    "return 0; " +

    "else " +

    "redis.call('del', KEYS[1]); " +

    "redis.call('publish', KEYS[2], ARGV[1]); " +

    "return 1; "+

    "end; " +

    "return nil;",

    Arrays.asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));

    }

    Redission的unlock解锁也是基于Lua脚本实现的,内部逻辑是先判断锁是否存在,不存在说明已经被释放了,发布锁释放消息后返回,锁存在再判断当前线程是否锁拥有者,不是的话,无权释放返回,解锁的话,会减去重入的次数,重新更新过期时间,若重入数捡完,删除当前key,发布锁释放消息

    如果你现在在JAVA这条路上挣扎,也想在IT行业拿高薪,可以参加我们的训

    练营课程,选择最适合自己的课程学习,技术大牛亲授,7个月后,进入名

    企拿高薪。我们的课程内容有:Java工程化、高性能及分布式、高性能、高

    架构、性能调优、Spring,MyBatis,Netty源码分析和大数据等多个知识点

    。如果你想拿高薪的,想学习的,想就业前景好的,想跟别人竞争能取得优

    势的,想进阿里面试但担心面试不过的,你都可以来,q群号为:835638062

    注:加群要求

    1、具有1-5工作经验的,面对目前流行的技术不知从何下手,需要突破技术

    瓶颈的可以加。

    2、在公司待久了,过得很安逸,但跳槽时面试碰壁。需要在短时间内进修

    、跳槽拿高薪的可以加。

    3、如果没有工作经验,但基础非常扎实,对java工作机制,常用设计思想

    ,常用java开发框架掌握熟练的,可以加。

    4、觉得自己很牛B,一般需求都能搞定。但是所学的知识点没有系统化,很

    难在技术领域继续突破的可以加。

    5.阿里Java高级大牛直播讲解知识点,分享知识,多年工作经验的梳理和总

    结,带着大家全面、科学地建立自己的技术体系和技术认知!

    相关文章

      网友评论

        本文标题:分布式锁实现:Redis

        本文链接:https://www.haomeiwen.com/subject/vmvyiftx.html