分布式锁用Redis好？还是Zookeeper好？-低调大师

分布式锁用Redis好？还是Zookeeper好？

2020-12-29 547

图片来自 Pexels

不过目前互联网项目越来越多的项目采用集群部署，也就是分布式情况，这两种锁就有些不够用了。

来两张图举例说明下，本地锁的情况下：

分布式锁情况下：

就其思想来说，就是一种“我全都要”的思想，所有服务都到一个统一的地方来取锁，只有取到锁的才能继续执行下去。

说完思想，下面来说一下具体的实现。

Redis 实现

为实现分布式锁，在 Redis 中存在 SETNX key value 命令，意为 set if not exists（如果不存在该 key，才去 set 值），就比如说是张三去上厕所，看厕所门锁着，他就不进去了，厕所门开着他才去。

可以看到，第一次 set 返回了 1，表示成功，但是第二次返回 0，表示 set 失败，因为已经存在这个 key 了。

当然只靠 setnx 这个命令可以吗？当然是不行的，试想一种情况，张三在厕所里，但他在里面一直没有释放，一直在里面蹲着，那外面人想去厕所全部都去不了，都想锤死他了。

Redis 同理，假设已经进行了加锁，但是因为宕机或者出现异常未释放锁，就造成了所谓的“死锁”。

聪明的你们肯定早都想到了，为它设置过期时间不就好了，可以 SETEX key seconds value 命令，为指定 key 设置过期时间，单位为秒。

但这样又有另一个问题，我刚加锁成功，还没设置过期时间，Redis 宕机了不就又死锁了，所以说要保证原子性吖，要么一起成功，要么一起失败。

当然我们能想到的 Redis 肯定早都为你实现好了，在 Redis 2.8 的版本后，Redis 就为我们提供了一条组合命令 SET key value ex seconds nx，加锁的同时设置过期时间。

就好比是公司规定每人最多只能在厕所呆 2 分钟，不管释放没释放完都得出来，这样就解决了“死锁”问题。

但这样就没有问题了吗？怎么可能。

试想又一种情况，厕所门肯定只能从里面开啊，张三上完厕所后张四进去锁上门，但是外面人以为还是张三在里面，而且已经过了 3 分钟了，就直接把门给撬开了，一看里面却是张四，这就很尴尬啊。

换成 Redis 就是说比如一个业务执行时间很长，锁已经自己过期了，别人已经设置了新的锁，但是当业务执行完之后直接释放锁，就有可能是删除了别人加的锁，这不是乱套了吗。

所以在加锁时候，要设一个随机值，在删除锁时进行比对，如果是自己的锁，才删除。

多说无益，烦人，直接上代码：

//基于jedis和lua脚本来实现
privatestaticfinal String LOCK_SUCCESS = "OK";
privatestaticfinal Long RELEASE_SUCCESS = 1L;
privatestaticfinal String SET_IF_NOT_EXIST = "NX";
privatestaticfinal String SET_WITH_EXPIRE_TIME = "PX";

@Override
public String acquire() {
    try {
        // 获取锁的超时时间，超过这个时间则放弃获取锁
        long end = System.currentTimeMillis() + acquireTimeout;
        // 随机生成一个 value
        String requireToken = UUID.randomUUID().toString();
        while (System.currentTimeMillis() < end) {
            String result = jedis
                .set(lockKey, requireToken, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
            if (LOCK_SUCCESS.equals(result)) {
                return requireToken;
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    } catch (Exception e) {
        log.error("acquire lock due to error", e);
    }

    returnnull;
}

@Override
public boolean release(String identify) {
    if (identify == null) {
        returnfalse;
    }
    //通过lua脚本进行比对删除操作，保证原子性
    String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    Object result = new Object();
    try {
        result = jedis.eval(script, Collections.singletonList(lockKey),
            Collections.singletonList(identify));
        if (RELEASE_SUCCESS.equals(result)) {
            log.info("release lock success, requestToken:{}", identify);
            returntrue;
        }
    } catch (Exception e) {
        log.error("release lock due to error", e);
    } finally {
        if (jedis != null) {
            jedis.close();
        }
    }

    log.info("release lock failed, requestToken:{}, result:{}", identify, result);
    returnfalse;
}

思考：加锁和释放锁的原子性可以用 lua 脚本来保证，那锁的自动续期改如何实现呢？

Redisson 实现

Redisson 顾名思义，Redis 的儿子，本质上还是 Redis 加锁，不过是对 Redis 做了很多封装，它不仅提供了一系列的分布式的 Java 常用对象，还提供了许多分布式服务。

在引入 Redisson 的依赖后，就可以直接进行调用：

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.4</version>
</dependency>

先来一段 Redisson 的加锁代码：

private void test() {
    //分布式锁名  锁的粒度越细，性能越好
    RLock lock = redissonClient.getLock("test_lock");
    lock.lock();
    try {
        //具体业务......
    } finally {
        lock.unlock();
    }
}

就是这么简单，使用方法 jdk 的 ReentrantLock 差不多，并且也支持 ReadWriteLock（读写锁）、Reentrant Lock（可重入锁）、Fair Lock（公平锁）、RedLock（红锁）等各种锁，详细可以参照redisson官方文档来查看。

那么 Redisson 到底有哪些优势呢？锁的自动续期（默认都是 30 秒），如果业务超长，运行期间会自动给锁续上新的 30s，不用担心业务执行时间超长而锁被自动删掉。

加锁的业务只要运行完成，就不会给当前续期，即便不手动解锁，锁默认在 30s 后删除，不会造成死锁问题。

前面也提到了锁的自动续期，我们来看看 Redisson 是如何来实现的。

先说明一下，这里主要讲的是 Redisson 中的 RLock，也就是可重入锁，有两种实现方法：

// 最常见的使用方法
lock.lock();

// 加锁以后10秒钟自动解锁
// 无需调用unlock方法手动解锁
lock.lock(10, TimeUnit.SECONDS);

而只有无参的方法是提供锁的自动续期操作的，内部使用的是“看门狗”机制，我们来看一看源码。

不管是空参还是带参方法，它们都调用的是同一个 lock 方法，未传参的话时间传了一个 -1，而带参的方法传过去的就是实际传入的时间。

继续点进 scheduleExpirationRenewal 方法：

点进 renewExpiration 方法：

总结一下，就是当我们指定锁过期时间，那么锁到时间就会自动释放。如果没有指定锁过期时间，就使用看门狗的默认时间 30s，只要占锁成功，就会启动一个定时任务，每隔 10s 给锁设置新的过期时间，时间为看门狗的默认时间，直到锁释放。

小结：虽然 lock() 有自动续锁机制，但是开发中还是推荐使用 lock(time，timeUnit)，因为它省掉了整个续期带来的性能损，可以设置过期时间长一点，搭配 unlock()。

若业务执行完成，会手动释放锁，若是业务执行超时，那一般我们服务也都会设置业务超时时间，就直接报错了，报错后就会通过设置的过期时间来释放锁。

public void test() {
    RLock lock = redissonClient.getLock("test_lock");
    lock.lock(30, TimeUnit.SECONDS);
    try {
        //.......具体业务
    } finally {
        //手动释放锁
        lock.unlock();
    }
}

基于 Zookeeper 来实现分布式锁

很多小伙伴都知道在分布式系统中，可以用 ZK 来做注册中心，但其实在除了做祖册中心以外，用 ZK 来做分布式锁也是很常见的一种方案。

先来看一下 ZK 中是如何创建一个节点的？ZK 中存在 create [-s] [-e] path [data] 命令，-s 为创建有序节点，-e 创建临时节点。

这样就创建了一个父节点并为父节点创建了一个子节点，组合命令意为创建一个临时的有序节点。

而 ZK 中分布式锁主要就是靠创建临时的顺序节点来实现的。至于为什么要用顺序节点和为什么用临时节点不用持久节点？先考虑一下，下文将作出说明。

同时还有 ZK 中如何查看节点？ ZK 中 ls [-w] path 为查看节点命令，-w 为添加一个 watch（监视器），/ 为查看根节点所有节点，可以看到我们刚才所创建的节点，同时如果是跟着指定节点名字的话为查看指定节点下的子节点。

后面的 00000000 为 ZK 为顺序节点增加的顺序。注册监听器也是 ZK 实现分布式锁中比较重要的一个东西。

下面来看一下 ZK 实现分布式锁的主要流程：

当第一个线程进来时会去父节点上创建一个临时的顺序节点。
第二个线程进来发现锁已经被持有了，就会为当前持有锁的节点注册一个 watcher 监听器。
第三个线程进来发现锁已经被持有了，因为是顺序节点的缘故，就会为上一个节点去创建一个 watcher 监听器。
当第一个线程释放锁后，删除节点，由它的下一个节点去占有锁。

看到这里，聪明的小伙伴们都已经看出来顺序节点的好处了。非顺序节点的话，每进来一个线程进来都会去持有锁的节点上注册一个监听器，容易引发“羊群效应”。

这么大一群羊一起向你飞奔而来，不管你顶不顶得住，反正 ZK 服务器是会增大宕机的风险。

而顺序节点的话就不会，顺序节点当发现已经有线程持有锁后，会向它的上一个节点注册一个监听器，这样当持有锁的节点释放后，也只有持有锁的下一个节点可以抢到锁，相当于是排好队来执行的，降低服务器宕机风险。

至于为什么使用临时节点，和 Redis 的过期时间一个道理，就算 ZK 服务器宕机，临时节点会随着服务器的宕机而消失，避免了死锁的情况。

下面来上一段代码的实现：

public class ZooKeeperDistributedLock implements Watcher {

    private ZooKeeper zk;
    private String locksRoot = "/locks";
    private String productId;
    private String waitNode;
    private String lockNode;
    private CountDownLatch latch;
    private CountDownLatch connectedLatch = new CountDownLatch(1);
    private int sessionTimeout = 30000;

    public ZooKeeperDistributedLock(String productId) {
        this.productId = productId;
        try {
            String address = "192.168.189.131:2181,192.168.189.132:2181";
            zk = new ZooKeeper(address, sessionTimeout, this);
            connectedLatch.await();
        } catch (IOException e) {
            throw new LockException(e);
        } catch (KeeperException e) {
            throw new LockException(e);
        } catch (InterruptedException e) {
            throw new LockException(e);
        }
    }

    public void process(WatchedEvent event) {
        if (event.getState() == KeeperState.SyncConnected) {
            connectedLatch.countDown();
            return;
        }

        if (this.latch != null) {
            this.latch.countDown();
        }
    }

    public void acquireDistributedLock() {
        try {
            if (this.tryLock()) {
                return;
            } else {
                waitForLock(waitNode, sessionTimeout);
            }
        } catch (KeeperException e) {
            throw new LockException(e);
        } catch (InterruptedException e) {
            throw new LockException(e);
        }
    }
    //获取锁
    public boolean tryLock() {
        try {
        // 传入进去的locksRoot + “/” + productId
        // 假设productId代表了一个商品id，比如说1
        // locksRoot = locks
        // /locks/10000000000，/locks/10000000001，/locks/10000000002
        lockNode = zk.create(locksRoot + "/" + productId, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        // 看看刚创建的节点是不是最小的节点
        // locks：10000000000，10000000001，10000000002
        List<String> locks = zk.getChildren(locksRoot, false);
        Collections.sort(locks);

        if(lockNode.equals(locksRoot+"/"+ locks.get(0))){
            //如果是最小的节点,则表示取得锁
            return true;
        }

        //如果不是最小的节点，找到比自己小1的节点
      int previousLockIndex = -1;
            for(int i = 0; i < locks.size(); i++) {
        if(lockNode.equals(locksRoot + “/” + locks.get(i))) {
                    previousLockIndex = i - 1;
            break;
        }
       }

       this.waitNode = locks.get(previousLockIndex);
        } catch (KeeperException e) {
            throw new LockException(e);
        } catch (InterruptedException e) {
            throw new LockException(e);
        }
        return false;
    }

    private boolean waitForLock(String waitNode, long waitTime) throws InterruptedException, KeeperException {
        Stat stat = zk.exists(locksRoot + "/" + waitNode, true);
        if (stat != null) {
            this.latch = new CountDownLatch(1);
            this.latch.await(waitTime, TimeUnit.MILLISECONDS);
            this.latch = null;
        }
        return true;
    }

    //释放锁
    public void unlock() {
        try {
            System.out.println("unlock " + lockNode);
            zk.delete(lockNode, -1);
            lockNode = null;
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }
    //异常
    public class LockException extends RuntimeException {
        private static final long serialVersionUID = 1L;

        public LockException(String e) {
            super(e);
        }

        public LockException(Exception e) {
            super(e);
        }
    }
}

总结

既然明白了 Redis 和 ZK 分别对分布式锁的实现，那么总该有所不同的吧。没错，我都帮大家整理好了：

实现方式的不同，Redis 实现为去插入一条占位数据，而 ZK 实现为去注册一个临时节点。
遇到宕机情况时，Redis 需要等到过期时间到了后自动释放锁，而 ZK 因为是临时节点，在宕机时候已经是删除了节点去释放锁。
Redis 在没抢占到锁的情况下一般会去自旋获取锁，比较浪费性能，而 ZK 是通过注册监听器的方式获取锁，性能而言优于 Redis。

不过具体要采用哪种实现方式，还是需要具体情况具体分析，结合项目引用的技术栈来落地实现。

作者：whynot_0

出处：juejin.im/post/6891571079702118407

程序员过关斩将--少年派登录安全的奇幻遐想

程序员过关斩将--并发控制中的一个小提醒

程序员过关斩将--请不要误会redis 6.0 的多线程

本文分享自微信公众号 - 架构师修行之路（jiagoushixiuxing）。
如有侵权，请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。

微信关注我们

原文链接：https://my.oschina.net/caicaijun/blog/4869647

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

JS必知的6种继承方式

JS作为面向对象的弱类型语言，继承也是其非常强大的特性之一。那么如何在JS中实现继承呢？让我们拭目以待 JS继承的实现方式既然要实现继承，那么首先我们得有一个父类，代码如下： //父类 functionPerson(name){//给构造函数添加了参数 this.name=name; this.sum=function(){ alert(this.name) } } Person.prototype.age=10;//给构造函数添加了原型属性 1、原型链继承 //原型链继承 functionPer(){ this.name="ker"; } Per.prototype=newPerson();//主要 varper1=newPer(); console.log(per1.age);//10 //instanceof判断元素是否在另一个元素的原型链上 //per1继承了Person的属性，返回true console.log(per1instanceofPerson);//true 重点：让新实例的原型等于父类的实例特点：实例可继承的属性有：实例...

2020-12-30

530

写在前面最近在推产品SaaS化的工作，以前虽然用过很多SaaS产品，但系统落地过程的经验还是空白，而且SaaS某种程度是产品完整性的解决方案，技术维度的挑战其实不是最大的。如何产品化，如何模块化这些才是重点。做好一个大型系统的架构很重要的一点是要做行研，看看业界大家怎么做，我一般会看有赞和京东，顺道买了点有赞的股票。 SaaS服务平台今天先简要总结下对于京东的一个SaaS平台的了解，其实SaaS上次可支撑的业务有很多，比如面向C端的娱乐工具产品，面向B端的运营平台，还有面对开发者的服务平台。京东的这款SaaS是专注于产业数智化的解决方案，通过京东多年沉淀下来的零售领域技术及方法论，集成并打通了技术、业务、用户、数据的四大能力，满足客户多业态多模式的场景化需求。其实上面这段可以忽略，任何一个面向B端的SaaS产品都是这么说的。挑战先说下一个服务平台类的SaaS产品的挑战会有哪些呢？从开发者角度来说，SaaS化产品均需要账号、员工、权限、审批流等通用服务，独立开发成本高。但是SaaS本身的商业能力体现确是千差万别的，比如有不同的定价、订购、履约的流程。有的时候需要和客服存量系...

2020-12-30

495

资源下载

更多资源

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。