第三章_基于zookeeper实现分布式锁

news2024/11/20 10:30:12

实现分布式锁目前有三种流行方案，分别为基于数据库、Redis、Zookeeper的方案。这里主要介绍基于zk怎么实现分布式锁。在实现分布式锁之前，先回顾zookeeper的知识点。

知识点回顾

Zookeeper（业界简称zk）是一种提供配置管理、分布式协同以及命名的中心化服务，这些提供的
功能都是分布式系统中非常底层且必不可少的基本功能，但是如果自己实现这些功能而且要达到高吞吐、低延迟同时还要保持一致性和可用性，实际上非常困难。因此zookeeper提供了这些功能，开发者在zookeeper之上构建自己的各种分布式系统。 

相关概念

Zookeeper提供一个多层级的节点命名空间（节点称为znode），每个节点都用一个以斜杠（/）分隔的路径表示，而且每个节点都有父节点（根节点除外），非常类似于文件系统。并且每个节点都是唯一的。

znode节点有四种类型：

  • PERSISTENT：永久节点。客户端与zookeeper断开连接后，该节点依旧存在。
  • EPHEMERAL：临时节点。客户端与zookeeper断开连接后，该节点被删除。
  • PERSISTENT_SEQUENTIAL：永久节点、序列化。客户端与zookeeper断开连接后，该节点依旧存在，只是Zookeeper给该节点名称进行顺序编号。
  • EPHEMERAL_SEQUENTIAL：临时节点、序列化。客户端与zookeeper断开连接后，该节点被删除，只是Zookeeper给该节点名称进行顺序编号。

创建这四种节点：

事件监听：在读取数据时，我们可以同时对节点设置事件监听，当节点数据或结构变化时，zookeeper会通知客户端。当前zookeeper有如下四种事件： 

  1. 节点创建
  2. 节点删除
  3. 节点数据修改
  4. 子节点变更

 java客户端

1. 引入依赖

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.14</version>
</dependency>

2. 常用api及其方法

// 初始化zookeeper客户端类，负责建立与zkServer的会话
new ZooKeeper(connectString, 30000, new Watcher() {
    @Override
    public void process(WatchedEvent event) {
        System.out.println("获取链接成功！！");
    }
});

// 创建一个节点，1-节点路径 2-节点内容 3-访问控制控制 4-节点类型
String fullPath = zooKeeper.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);

// 判断一个节点是否存在
Stat stat = zooKeeper.exists(rootPath, false);
if(stat != null){...}

// 查询一个节点的内容
Stat stat = new Stat();
byte[] data = zooKeeper.getData(path, false, stat);

// 更新一个节点
zooKeeper.setData(rootPath, new byte[]{}, stat.getVersion() + 1);
// 删除一个节点
zooKeeper.delete(path, stat.getVersion());

// 查询一个节点的子节点列表
List<String> children = zooKeeper.getChildren(rootPath, false);

// 关闭链接
if (zooKeeper != null){ zooKeeper.close(); }

思路分析

分布式锁的步骤：

  1. 获取锁：create一个节点
  2. 删除锁：delete一个节点
  3. 重试：没有获取到锁的请求重试 

参照redis分布式锁的特点：

  1. 互斥 排他。
  2. 防死锁：

    1. 可自动释放锁（临时节点） ：获得锁之后客户端所在机器宕机了，客户端没有主动删除子节点；如果创建的是永久的节点，那么这个锁永远不会释放，导致死锁；由于创建的是临时节点，客户端宕机后，过了一定时间zookeeper没有收到客户端的心跳包判断会话失效，将临时节点删除从而释放锁。

    2. 可重入锁：借助于ThreadLocal。
     
  3. 防误删：宕机自动释放临时节点，不需要设置过期时间，也就不存在误删问题。
  4. 加锁/解锁要具备原子性。
  5. 单点问题：使用Zookeeper可以有效的解决单点问题，ZK一般是集群部署的。
  6. 集群问题：zookeeper集群是强一致性的，只要集群中有半数以上的机器存活，就可以对外提供服务。

基本实现

实现思路：

  1. 多个请求同时添加一个相同的临时节点，只有一个可以添加成功。添加成功的获取到锁。
  2. 执行业务逻辑。
  3. 完成业务流程后，删除节点释放锁。

由于zookeeper获取链接是一个耗时过程，这里可以在项目启动时，初始化链接，并且只初始化一次。借助于spring特性，代码实现如下：

@Component
public class ZkClient {

    private static final String connectString = "172.16.116.100:2181";

    private static final String ROOT_PATH = "/distributed";

    private ZooKeeper zooKeeper;

    @PostConstruct
    public void init() {
        try {
            // 连接zookeeper服务器
            this.zooKeeper = new ZooKeeper(connectString, 30000, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    System.out.println("获取链接成功！！");
                }
            });

            // 创建分布式锁根节点
            if (this.zooKeeper.exists(ROOT_PATH, false) == null) {
                this.zooKeeper.create(ROOT_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (Exception e) {
            System.out.println("获取链接失败！");
            e.printStackTrace();
        }
    }

    @PreDestroy
    public void destroy() {
        try {
            if (zooKeeper != null) {
                zooKeeper.close();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 初始化zk分布式锁对象方法
     * @param lockName 所名称
     * @return zk分布式锁对象
     */
    public ZkDistributedLock getZkDistributedLock(String lockName) {
        return new ZkDistributedLock(zooKeeper, lockName);
    }

}

zk分布式锁具体实现：

public class ZkDistributedLock {

    private static final String ROOT_PATH = "/distributed";

    private String path;

    private ZooKeeper zooKeeper;

    public ZkDistributedLock(ZooKeeper zooKeeper, String lockName) {
        this.zooKeeper = zooKeeper;
        this.path = ROOT_PATH + "/" + lockName;
    }

    public void lock() {
        try {
            zooKeeper.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        } catch (Exception e) {
            // 重试
            try {
                Thread.sleep(200);
                lock();
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
        }
    }

    public void unlock() {
        try {
            this.zooKeeper.delete(path, 0);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }
}

改造上一章redis分布式锁中StockService的checkAndLock方法：

@Autowired
private ZkClient client;

public void checkAndLock() {
    // 加锁，获取锁失败重试
    ZkDistributedLock lock = this.client.getZkDistributedLock("lock");
    lock.lock();

    // 先查询库存是否充足
    Stock stock = this.stockMapper.selectById(1L);

    // 再减库存
    if (stock != null && stock.getCount() > 0) {
        stock.setCount(stock.getCount() - 1);
        this.stockMapper.updateById(stock);
    }

    // 释放锁
    lock.unlock();
}

Jmeter压力测试：

性能一般，mysql数据库的库存余量为0（注意：所有测试之前都要先修改库存量为5000）

基本实现存在的问题：

  1. 性能一般（比mysql略好）
  2. 不可重入

接下来首先来提高性能

优化：性能优化

基本实现中由于无限自旋影响性能：

试想：每个请求要想正常的执行完成，最终都是要创建节点，如果能够避免争抢必然可以提高性能。这里借助于zk的临时序列化节点，实现分布式锁： 

实现阻塞锁 

代码实现：

public class ZkDistributedLock {

    private static final String ROOT_PATH = "/distributed";

    private String path;

    private ZooKeeper zooKeeper;

    public ZkDistributedLock(ZooKeeper zooKeeper, String lockName) {
        try {
            this.zooKeeper = zooKeeper;
            this.path = zooKeeper.create(ROOT_PATH + "/" + lockName + "-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void lock() {
        String preNode = getPreNode(path);

        // 如果该节点没有前一个节点，说明该节点时最小节点，放行执行业务逻辑
        if (StringUtils.isEmpty(preNode)) {
            return;
        }

        // 重新检查。是否获取到锁
        try {
            Thread.sleep(20);
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }

        lock();
    }

    public void unlock() {
        try {
            this.zooKeeper.delete(path, 0);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取指定节点的前节点
     * @param path 路径
     * @return 前节点
     */
    private String getPreNode(String path) {
        try {
            // 获取当前节点的序列化号
            Long curSerial = Long.valueOf(StringUtils.substringAfterLast(path, "-"));
            // 获取根路径下的所有序列化子节点
            List<String> nodes = this.zooKeeper.getChildren(ROOT_PATH, false);

            // 判空
            if (CollectionUtils.isEmpty(nodes)) {
                return null;
            }

            // 获取前一个节点
            Long flag = 0L;
            String preNode = null;

            for (String node : nodes) {
                // 获取每个节点的序列化号
                Long serial = Long.valueOf(StringUtils.substringAfterLast(node, "-"));
                if (serial < curSerial && serial > flag) {
                    flag = serial;
                    preNode = node;
                }
            }

            return preNode;
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return null;
    }
}

主要修改了构造方法和lock方法：

并添加了getPreNode获取前置节点的方法。

测试结果如下：

性能反而更弱了。

原因：虽然不用反复争抢创建节点了，但是会自选判断自己是最小的节点，这个判断逻辑反而更复杂更耗时。

解决方案：监听。

监听实现阻塞锁

对于这个算法有个极大的优化点：假如当前有1000个节点在等待锁，如果获得锁的客户端释放锁时，这1000个客户端都会被唤醒，这种情况称为“羊群效应”；在这种羊群效应中，zookeeper需要通知1000个客户端，这会阻塞其他的操作，最好的情况应该只唤醒新的最小节点对应的客户端。应该怎么做呢？在设置事件监听时，每个客户端应该对刚好在它之前的子节点设置事件监听，例如子节点列表为/lock/lock-0000000000、/lock/lock-0000000001、/lock/lock-0000000002，序号为1的客户端监听序号为0的子节点删除消息，序号为2的监听序号为1的子节点删除消息。

所以调整后的分布式锁算法流程如下：

  • 客户端连接zookeeper，并在/lock下创建临时的且有序的子节点，第一个客户端对应的子节点为/lock/lock-0000000000，第二个为/lock/lock-0000000001，以此类推；
  • 客户端获取/lock下的子节点列表，判断自己创建的子节点是否为当前子节点列表中序号最小的子节点，如果是则认为获得锁，否则监听刚好在自己之前一位的子节点删除消息，获得子节点变更通知后重复此步骤直至获得锁；
  • 执行业务代码；
  • 完成业务流程后，删除对应的子节点释放锁。

改造ZkDistributedLock的lock方法：

public void lock() {
    try {
        String preNode = getPreNode(path);

        // 如果该节点没有前一个节点，说明该节点时最小节点，放行执行业务逻辑
        if (StringUtils.isEmpty(preNode)) {
            return;
        } else {
            CountDownLatch countDownLatch = new CountDownLatch(1);

            if (this.zooKeeper.exists(ROOT_PATH + "/" + preNode, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    countDownLatch.countDown();
                }
            }) == null) {
                return;
            }

            // 阻塞。。。。
            countDownLatch.await();
            return;
        }
    } catch (Exception e) {
        e.printStackTrace();
        // 重新检查。是否获取到锁
        try {
            Thread.sleep(200);
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }

        lock();
    }
}

 压力测试效果如下：

由此可见性能提高不少仅次于redis的分布式锁。

优化：可重入锁

引入ThreadLocal线程局部变量保证zk分布式锁的可重入性。

public class ZkDistributedLock {

    private static final String ROOT_PATH = "/distributed";

    private static final ThreadLocal<Integer> THREAD_LOCAL = new ThreadLocal<>();

    private String path;

    private ZooKeeper zooKeeper;

    public ZkDistributedLock(ZooKeeper zooKeeper, String lockName) {
        try {
            this.zooKeeper = zooKeeper;

            if (THREAD_LOCAL.get() == null || THREAD_LOCAL.get() == 0){
                this.path = zooKeeper.create(ROOT_PATH + "/" + lockName + "-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void lock() {
        Integer flag = THREAD_LOCAL.get();

        if (flag != null && flag > 0) {
            THREAD_LOCAL.set(flag + 1);
            return;
        }

        try {
            String preNode = getPreNode(path);

            // 如果该节点没有前一个节点，说明该节点时最小节点，放行执行业务逻辑
            if (StringUtils.isEmpty(preNode)) {
                THREAD_LOCAL.set(1);
                return ;
            } else {
                CountDownLatch countDownLatch = new CountDownLatch(1);

                if (this.zooKeeper.exists(ROOT_PATH + "/" + preNode, new Watcher() {
                    @Override
                    public void process(WatchedEvent event) {
                        countDownLatch.countDown();
                    }
                }) == null) {
                    THREAD_LOCAL.set(1);
                    return;
                }

                // 阻塞。。。。
                countDownLatch.await();
                THREAD_LOCAL.set(1);
                return;
            }
        } catch (Exception e) {
            e.printStackTrace();
            // 重新检查。是否获取到锁
            try {
                Thread.sleep(200);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }

            lock();
        }
    }

    public void unlock() {
        try {
            THREAD_LOCAL.set(THREAD_LOCAL.get() - 1);

            if (THREAD_LOCAL.get() == 0) {
                this.zooKeeper.delete(path, 0);
                THREAD_LOCAL.remove();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取指定节点的前节点
     * @param path 路径
     * @return 前节点
     */
    private String getPreNode(String path) {
        try {
            // 获取当前节点的序列化号
            Long curSerial =
            Long.valueOf(StringUtils.substringAfterLast(path, "-"));
            // 获取根路径下的所有序列化子节点
            List<String> nodes = this.zooKeeper.getChildren(ROOT_PATH, false);

            // 判空
            if (CollectionUtils.isEmpty(nodes)) {
                return null;
            }

            // 获取前一个节点
            Long flag = 0L;
            String preNode = null;

            for (String node : nodes) {
                // 获取每个节点的序列化号
                Long serial = Long.valueOf(StringUtils.substringAfterLast(node, "-"));

                if (serial < curSerial && serial > flag) {
                    flag = serial;
                    preNode = node;
                }
            }

            return preNode;
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return null;
    }
}

 zk分布式锁小结

参照redis分布式锁的特点：

  1. 互斥 排他：zk节点的不可重复性，以及序列化节点的有序性。
  2. 防死锁：

    1. 可自动释放锁：临时节点
    2. 可重入锁：借助于ThreadLocal
     
  3. 防误删：临时节点。
  4. 加锁/解锁要具备原子性。
  5. 单点问题：使用Zookeeper可以有效的解决单点问题，ZK一般是集群部署的。
  6. 集群问题：zookeeper集群是强一致性的，只要集群中有半数以上的机器存活，就可以对外提供服务。
  7. 公平锁：有序性节点。

 Curator中的分布式锁

Curator是netflix公司开源的一套zookeeper客户端，目前是Apache的顶级项目。与Zookeeper提供的原生客户端相比，Curator的抽象层次更高，简化了Zookeeper客户端的开发量。Curator解决了很多zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册wathcer和NodeExistsException 异常等。

通过查看官方文档，可以发现Curator主要解决了三类问题：

  1. 封装ZooKeeper client与ZooKeeper server之间的连接处理。
  2. 提供了一套Fluent风格的操作API。
  3. 提供ZooKeeper各种应用场景(recipe， 比如：分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等)的抽象封装，这些实现都遵循了zk的最佳实践，并考虑了各种极端情况。

Curator由一系列的模块构成，对于一般开发者而言，常用的是curator-framework和curatorrecipes：

  • curator-framework：提供了常见的zk相关的底层操作。
  • curator-recipes：提供了一些zk的典型使用场景的参考。本节重点关注的分布式锁就是该包提供的。

引入依赖：

最新版本的curator 4.3.0支持zookeeper 3.4.x和3.5，但是需要注意curator传递进来的依赖，需要和实际服务器端使用的版本相符，以我们目前使用的zookeeper 3.4.14为例。

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>4.3.0</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
        </exclusion>
    </exclusions>
</dependency>

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>4.3.0</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
        </exclusion>
    </exclusions>
</dependency>

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.14</version>
</dependency>

可重入锁InterProcessMutex

Reentrant和JDK的ReentrantLock类似， 意味着同一个客户端在拥有锁的同时，可以多次获取，不会被阻塞。它是由类InterProcessMutex来实现。

// 常用构造方法
public InterProcessMutex(CuratorFramework client, String path)
// 获取锁
public void acquire();
// 带超时时间的可重入锁
public boolean acquire(long time, TimeUnit unit);
// 释放锁
public void release();

添加curator客户端配置：

@Configuration
public class ZkCuratorConfig {

    @Bean
    public CuratorFramework curatorFramework() {
        // 后台重试，每个1000ms重试一次，重试3次
        RetryPolicy retry = new ExponentialBackoffRetry(1000, 3);
        // 初始化CuratorFramework客户端，如果有多个zk地址，以逗号分割。
        CuratorFramework client = CuratorFrameworkFactory.newClient("172.16.116.100:2181", retry);
        client.start();
        return client;
    }

}

改造service测试方法：

@Autowired
private CuratorFramework curatorFramework;

public void checkAndLock() {
    try {
        // 加锁，获取锁失败重试
        InterProcessMutex mutex = new InterProcessMutex(curatorFramework, "/curator/lock");
        mutex.acquire();

        // 先查询库存是否充足
        Stock stock = this.stockMapper.selectById(1L);

        // 再减库存
        if (stock != null && stock.getCount() > 0) {
            stock.setCount(stock.getCount() - 1);
            this.stockMapper.updateById(stock);
        }

        // 释放锁
        mutex.release();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

注意：如想重入，则需要使用同一个InterProcessMutex对象。

压力测试结果：

不可重入锁InterProcessSemaphoreMutex 

具体实现：InterProcessSemaphoreMutex。与InterProcessMutex调用方法类似，区别在于该锁是不可重入的，在同一个线程中不可重入。

public InterProcessSemaphoreMutex(CuratorFramework client, String path);
public void acquire();
public boolean acquire(long time, TimeUnit unit);
public void release();

可重入读写锁InterProcessReadWriteLock

类似JDK的ReentrantReadWriteLock。一个拥有写锁的线程可重入读锁，但是读锁却不能进入写锁。这也意味着写锁可以降级成读锁。从读锁升级成写锁是不成的。主要实现类InterProcessReadWriteLock：

// 构造方法
public InterProcessReadWriteLock(CuratorFramework client, String basePath);
// 获取读锁对象
InterProcessMutex readLock();
// 获取写锁对象
InterProcessMutex writeLock();

联锁InterProcessMultiLock

Multi Shared Lock是一个锁的容器。当调用acquire， 所有的锁都会被acquire，如果请求失败，所有的锁都会被release。同样调用release时所有的锁都被release(失败被忽略)。基本上，它就是组锁的代表，在它上面的请求释放操作都会传递给它包含的所有的锁。实现类InterProcessMultiLock：

// 构造函数需要包含的锁的集合，或者一组ZooKeeper的path
public InterProcessMultiLock(List<InterProcessLock> locks);
public InterProcessMultiLock(CuratorFramework client, List<String> paths);
// 获取锁
public void acquire();
public boolean acquire(long time, TimeUnit unit);
// 释放锁
public synchronized void release();

 信号量InterProcessSemaphoreV2

一个计数的信号量类似JDK的Semaphore。JDK中Semaphore维护的一组许可(permits)，而Cubator中称之为租约(Lease)。注意，所有的实例必须使用相同的numberOfLeases值。调用acquire会返回一个租约对象。客户端必须在finally中close这些租约对象，否则这些租约会丢失掉。但是，如果客户端session由于某种原因比如crash丢掉， 那么这些客户端持有的租约会自动close， 这样其它客户端可以继续使用这些租约。主要实现类InterProcessSemaphoreV2：

// 构造方法
public InterProcessSemaphoreV2(CuratorFramework client, String path, int maxLeases);

// 注意一次你可以请求多个租约，如果Semaphore当前的租约不够，则请求线程会被阻塞。
// 同时还提供了超时的重载方法
public Lease acquire();
public Collection<Lease> acquire(int qty);
public Lease acquire(long time, TimeUnit unit);
public Collection<Lease> acquire(int qty, long time, TimeUnit unit);

// 租约还可以通过下面的方式返还
public void returnAll(Collection<Lease> leases);
public void returnLease(Lease lease);

栅栏barrier

  1. DistributedBarrier构造函数中barrierPath参数用来确定一个栅栏，只要barrierPath参数相同(路径相同)就是同一个栅栏。通常情况下栅栏的使用如下：
     
    1. 主 client 设置一个栅栏
    2. 其他客户端就会调用 waitOnBarrier() 等待栅栏移除，程序处理线程阻塞
    3. 主 client移除栅栏，其他客户端的处理程序就会同时继续运行。

    DistributedBarrier类的主要方法如下：
      
    setBarrier() - 设置栅栏
waitOnBarrier() - 等待栅栏移除
removeBarrier() - 移除栅栏
  2. DistributedDoubleBarrier双栅栏，允许客户端在计算的开始和结束时同步。当足够的进程加入到双栅栏时，进程开始计算，当计算完成时，离开栅栏。DistributedDoubleBarrier实现了双栅栏的功能。构造函数如下：
      
    // client - the client
// barrierPath - path to use
// memberQty - the number of members in the barrier
public DistributedDoubleBarrier(CuratorFramework client, String barrierPath, int memberQty);
enter()、enter(long maxWait, TimeUnit unit) - 等待同时进入栅栏
leave()、leave(long maxWait, TimeUnit unit) - 等待同时离开栅栏

    memberQty是成员数量，当enter方法被调用时，成员被阻塞，直到所有的成员都调用了enter。当leave方法被调用时，它也阻塞调用线程，直到所有的成员都调用了leave。

    注意：参数memberQty的值只是一个阈值，而不是一个限制值。当等待栅栏的数量大于或等于这个值栅栏就会打开！

    与栅栏(DistributedBarrier)一样,双栅栏的barrierPath参数也是用来确定是否是同一个栅栏的，双栅栏的使用情况如下：

    1. 从多个客户端在同一个路径上创建双栅栏(DistributedDoubleBarrier),然后调用enter()方
    法，等待栅栏数量达到memberQty时就可以进入栅栏。
    2. 栅栏数量达到memberQty，多个客户端同时停止阻塞继续运行，直到执行leave()方法，等待memberQty个数量的栅栏同时阻塞到leave()方法中。
    3. memberQty个数量的栅栏同时阻塞到leave()方法中，多个客户端的leave()方法停止阻塞，继续运行。

倒计数器

利用ZooKeeper可以实现一个集群共享的计数器。只要使用相同的path就可以得到最新的计数器值，这是由ZooKeeper的一致性保证的。Curator有两个计数器， 一个是用int来计数，一个用long来计数。

SharedCount

这个类使用int类型来计数。主要涉及三个类。

* SharedCount
* SharedCountReader
* SharedCountListener

SharedCount代表计数器， 可以为它增加一个SharedCountListener，当计数器改变时此Listener可以监听到改变的事件，而SharedCountReader可以读取到最新的值， 包括字面值和带版本信息的值VersionedValue。

DistributedAtomicLong

除了计数的范围比SharedCount大了之外， 它首先尝试使用乐观锁的方式设置计数器， 如果不成功(比如期间计数器已经被其它client更新了)， 它使用InterProcessMutex方式来更新计数值。此计数器有一系列的操作：

  • get(): 获取当前值。
  • increment()：加一。
  • decrement(): 减一。
  • add()：增加特定的值。
  • subtract(): 减去特定的值。
  • trySet(): 尝试设置计数值。
  • forceSet(): 强制设置计数值。

你必须检查返回结果的succeeded()， 它代表此操作是否成功。如果操作成功， preValue()代表操作前的值， postValue()代表操作后的值。

总结

实现的复杂性或者难度角度：Zookeeper > 缓存 > 数据库
实际性能角度：缓存 > Zookeeper > 数据库
可靠性角度：Zookeeper > 缓存 > 数据库

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/653467.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

NIFI1.21.0_Mysql到Mysql增量CDC同步中_日期类型_以及null数据同步处理补充---大数据之Nifi工作笔记0057

NIFI1.21.0_Mysql到Mysql增量CDC同步中_日期类型_以及null数据同步处理补充---大数据之Nifi工作笔记0057

NIFI从MySql中增量同步数据_通过Mysql的binlog功能_实时同步mysql数据_根据binlog实现update数据实时同步_实际操作05---大数据之Nifi工作笔记0044 具体的,之前已经写过,如何在NIFI中实现MySQL的增量数据同步,但是写的简单了,因为,比如在插入的时候,更新的时候,仅仅是写死的某…
阅读更多...
第五节 利用Ogre 2.3实现雨，雪，爆炸，飞机喷气尾焰等粒子效果

第五节 利用Ogre 2.3实现雨，雪，爆炸，飞机喷气尾焰等粒子效果

本节主要学习如何使用Ogre2.3加载粒子效果。为了学习方便&#xff0c;直接将官方粒子模块Sample_ParticleFX单独拿出来编译&#xff0c;学习如何实现粒子效果。 一. 前提须知 如果参考官方示例建议用最新版的Ogre 2.3.1。否则找不到有粒子效果的示例。不要用官网Ogre2.3 scri…
阅读更多...
【微信小程序开发】第 8 课 - 小程序 API 的 3 大分类

【微信小程序开发】第 8 课 - 小程序 API 的 3 大分类

欢迎来到博主 Apeiron 的博客&#xff0c;祝您旅程愉快 &#xff01; 时止则止&#xff0c;时行则行。动静不失其时&#xff0c;其道光明。 目录 1、小程序 API 概述 2、小程序 API 的 3 大分类 3、总结 1、小程序 API 概述 小程序中的 API 是由宿主环境提供的&#xff0c;…
阅读更多...
一款基于 SpringCloud 的电商商城系统，小程序+管理端一套带走

一款基于 SpringCloud 的电商商城系统，小程序+管理端一套带走

项目介绍 Smart Shop 是一款基于 Spring Cloud MybatisPlusXXL-JOBredisVue 的前后端分离、分布式、微服务架构的 Java 商城系统&#xff0c;采用稳定框架开发及优化核心&#xff0c;减少依赖&#xff0c;具备出色的执行效率&#xff0c;扩展性、稳定性高&#xff0c;H5/小程序…
阅读更多...
pnpm + monorepo架构思想小试牛刀

pnpm + monorepo架构思想小试牛刀

写在前面 今天要写的是关于一种前端全新架构的方式&#xff0c;monorepo这是目前相对来讲比较新的一种前端架构&#xff0c;整好趁着最近在学&#xff0c;就利用这个平台记录一下学习的一个过程&#xff0c;上一篇文章更新的是react&#xff0c;后面也会一样更新&#xff0c;今…
阅读更多...
深入理解Java虚拟机jvm-运行时数据区域（基于OpenJDK12）

深入理解Java虚拟机jvm-运行时数据区域（基于OpenJDK12）

运行时数据区域 运行时数据区域程序计数器Java虚拟机栈本地方法栈Java堆方法区运行时常量池直接内存 运行时数据区域 Java虚拟机在执行Java程序的过程中会把它所管理的内存划分为若干个不同的数据区域。这些区域 有各自的用途&#xff0c;以及创建和销毁的时间&#xff0c;有的…
阅读更多...
持续改进与创新：水库大坝安全管理方式

持续改进与创新：水库大坝安全管理方式

随着工业的快速发展&#xff0c;大坝建设已成为经济发展的重要部分。然而&#xff0c;由于自然环境的破坏以及人类因素的干扰&#xff0c;大坝的安全问题备受关注。每年都有不少大坝事故爆发&#xff0c;造成无法预估的损失。据统计&#xff0c;截至2006年我国共有3260座水库已…
阅读更多...
【AntDB数据库】AntDB数据库跨地域多中心部署

【AntDB数据库】AntDB数据库跨地域多中心部署

跨地域多中心部署 **** 某省核心账务库案例 **** 通信行业核心业务系统已经与某款国外成熟商业数据库深度捆绑多年&#xff0c;为改变这一现状&#xff0c;实现数据库“自主可控”的目标&#xff0c;某省经过多轮调研选型与评测最终选择AntDB分布式内存数据库进行核心产生系统…
阅读更多...
Linux线程同步(上)

Linux线程同步(上)

文章目录 1. 同步的概念2. 条件变量函数2.1 等待函数2.2 样例 3. 生产者消费者模型4. 阻塞队列4.1 模拟阻塞队列的生产消费模型4.2 构造函数和析构函数4.3 生产接口和消费接口4.4 创建线程进行测试 1. 同步的概念 互斥可能会导致一个执行流长时间得不到某种资源。也叫饥饿问题…
阅读更多...
健身房小程序怎么做？

健身房小程序怎么做？

如果把预约小程序用好了&#xff0c;会给你的场馆经营带来很多意想不到的好处&#xff0c;解决用户线上约客的问题&#xff0c;顶多只发挥了 20% 的作用&#xff0c;那另外 80% 的用法是什么呢&#xff1f; 高效推荐名片服务行业做的是口碑&#xff0c;用户靠的是转介绍&#x…
阅读更多...
SignalR快速入门 ~ 仿QQ即时聊天，消息推送，单聊，群聊，多群公聊（基础=》提升）

SignalR快速入门 ~ 仿QQ即时聊天，消息推送，单聊，群聊，多群公聊（基础=》提升）

SignalR快速入门 ~ 仿QQ即时聊天&#xff0c;消息推送&#xff0c;单聊&#xff0c;群聊&#xff0c;多群公聊&#xff08;基础》提升&#xff0c;5个Demo贯彻全篇&#xff0c;感兴趣的玩才是真的学&#xff09; 官方demo:http://www.asp.net/signalr/overview/getting-started…
阅读更多...
超大模型如何实现3D WEB轻量化渲染？

超大模型如何实现3D WEB轻量化渲染？

Hoops Communicator是Tech Soft 3D旗下的主流产品之一&#xff0c;具有强大的、专用的高性能图形内核&#xff0c;专注于基于Web的高级3D工程应用程序。其由HOOPS Server和HOOPS Web Viewer两大部分组成&#xff0c;提供了HOOPS中的HOOPS Convertrer、Data Authoring的模型转换…
阅读更多...
电脑提示msvcr120.dll丢失的解决方法win10，总共有三种，那个更方便

电脑提示msvcr120.dll丢失的解决方法win10，总共有三种，那个更方便

电脑修复经验分享&#xff0c;dll动态链接库文件丢失修复教程&#xff0c;DLL 文件&#xff0c;即动态链接库&#xff0c;也有人称作应用程序拓展。一种可执行文件&#xff0c;允许程序共享执行特殊任务所需的代码和其他资源。msvcr120.dll也是属于dll文件之一&#xff0c;在Wi…
阅读更多...
python 面向对象 对象

python 面向对象 对象

类 构造函数 # 创建类 class Student:name None # 成员属性age None # 成员属性def say(self): # 成员方法print(self.name)# 构造函数def __init__(self,name,age):self.name nameself.age age#创建类对象 my_student Student() # 对象的属性 赋值 my_student.name …
阅读更多...
Mysql数据库——用户管理与授权

Mysql数据库——用户管理与授权

Mysql数据库——用户管理与授权 一、登录用户的管理1&#xff0e;新建用户2&#xff0e;查看用户信息3&#xff0e;重命名用户4&#xff0e;删除用户5&#xff0e;修改当前登录用户密码6&#xff0e;修改其他用户密码7&#xff0e;忘记 root 密码的解决办法 二、数据库用户授权…
阅读更多...
【Docker】Docker中Linux 容器、网络虚拟化与虚拟局域网的技术特点详细讲解

【Docker】Docker中Linux 容器、网络虚拟化与虚拟局域网的技术特点详细讲解

前言 Docker 是一个开源的应用容器引擎&#xff0c;让开发者可以打包他们的应用以及依赖包到一个可移植的容器中,然后发布到任何流行的Linux或Windows操作系统的机器上,也可以实现虚拟化,容器是完全使用沙箱机制,相互之间不会有任何接口。 &#x1f4d5;作者简介&#xff1a;热…
阅读更多...
Stream流学习前置知识之匿名内部类

Stream流学习前置知识之匿名内部类

目录 匿名内部类 概念 语法体 案例演示 匿名内部类 概念 没有名字的内部类 用于在程序中创建一个只使用一次的临时类 使用 new 关键字来创建一个对象&#xff0c;重写该类的方法或实现该类的接口 语法体 new 父类构造器或接口() {// 匿名内部类的定义 } 案例演示 T…
阅读更多...
【深度学习】基于Qt的人脸识别系统，门禁人脸识别系统，Python人脸识别流程，树莓派

【深度学习】基于Qt的人脸识别系统，门禁人脸识别系统，Python人脸识别流程，树莓派

文章目录 人脸识别过程人脸检测人脸对齐人脸特征提取特征距离比对人脸识别系统 人脸识别过程 在深度学习领域做人脸识别的识别准确率已经高到超出人类识别&#xff0c;但综合考虑模型复杂度&#xff08;推理速度&#xff09;和模型的识别效果&#xff0c;这个地方还是有做一些…
阅读更多...
网络安全大厂面试题合集+面试题文档

网络安全大厂面试题合集+面试题文档

以下为网络安全各个方向涉及的面试题&#xff0c;星数越多代表问题出现的几率越大&#xff0c;祝各位都能找到满意的工作。 注&#xff1a;本套面试题&#xff0c;已整理成pdf文档&#xff0c;但内容还在持续更新中&#xff0c;因为无论如何都不可能覆盖所有的面试问题&#xf…
阅读更多...
阿里巴巴开源Chat2DB v1.0.11 初体验

阿里巴巴开源Chat2DB v1.0.11 初体验

阿里巴巴开源Chat2DB v1.0.11 初体验 前言什么是Chat2DB下载安装安装配置Chat2DB初体验配置数据源准备测试数据认识几个功能菜单开始测试自然语言转SQLSQL解释SQL优化 使用总结后续功能结语 前言 作为一名阿里巴巴开源项目的拥护者&#xff0c;从Chat2DB开源至今都有关注这个开…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023