基于ZooKeeper的三种分布式锁实现

2019-08-26 06:12:57来源：博客园阅读 ()

基于ZooKeeper的三种分布式锁实现

【欢迎关注公众号：程序猿讲故事（codestory），及时接收最新文章】

今天介绍基于ZooKeeper的分布式锁的简单实现，包括阻塞锁和非阻塞锁。同时增加了网上很少介绍的基于节点的非阻塞锁实现，主要是为了加深对ZooKeeper的理解。

维基百科：分布式锁，是控制分布式系统之间同步访问共享资源的一种方式。在分布式系统中，常常需要协调他们的动作。如果不同的系统或是同一个系统的不同主机之间共享了一个或一组资源，那么访问这些资源的时候，往往需要互斥来防止彼此干扰来保证一致性，在这种情况下，便需要使用到分布式锁。

1 阻塞锁和非阻塞锁

根据业务特点，普通分布式锁有两种需求：阻塞锁和非阻塞锁。

阻塞锁：多个系统同时调用同一个资源，所有请求被排队处理。已经得到分布式锁的系统，进入运行状态完成业务操作；没有得到分布式锁的线程进入阻塞状态等待，当获得相应的信号并获得分布式锁后，进入运行状态完成业务操作。

非阻塞锁：多个系统同时调用同一个资源，当某一个系统最先获取到锁，进入运行状态完成业务操作；其他没有得到分布式锁的系统，就直接返回，不做任何业务逻辑，可以给用户提示进行其他操作。

2 锁代码简单设计

基于ZooKeeper实现锁，一般都是创建EPHEMERAL_SEQUENTIAL子节点并比较序号实现的。参照Redis的分布式锁实现，也可以使用EPHEMERAL节点实现。

3 分布式锁代码

完整代码比较多，占篇幅。在文中只保留了关键的代码。完整项目代码放到了github（https://github.com/SeemSilly/codestory/tree/master/research-zoo-keeper），感兴趣的可以关注。

3.1 分布式锁接口定义

ZooKeeperLock.java

public interface ZooKeeperLock {

  /**

   * 尝试获取锁

   *

   * @param guidNodeName 用于加锁的唯一节点名

   * @param clientGuid 用于唯一标识当前客户端的ID

   * @return

   */

  boolean lock(String guidNodeName, String clientGuid);

 

  /**

   * 释放锁

   *

   * @param guidNodeName 用于加锁的唯一节点名

   * @param clientGuid 用于唯一标识当前客户端的ID

   * @return

   */

  boolean release(String guidNodeName, String clientGuid);

 

  /**

   * 锁是否已经存在

   *

   * @param guidNodeName 用于加锁的唯一节点名

   * @return

   */

  boolean exists(String guidNodeName);

}

3.2 基于节点实现的非阻塞锁

NodeBlocklessLock.java

public class NodeBlocklessLock extends ZooKeeperBase implements ZooKeeperLock {

  /** 尝试获取锁 */

  public boolean lock(String guidNodeName, String clientGuid) {

    boolean result = false;

    if (getZooKeeper().exists(guidNodeName, false) == null) {

      getZooKeeper().create(guidNodeName, clientGuid.getBytes(),

          ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

      byte[] data = getZooKeeper().getData(guidNodeName, false, null);

      if (data != null && clientGuid.equals(new String(data))) {

        result = true;

      }

    }

    return result;

  }

 

  /** 释放锁 */

  public boolean release(String guidNodeName, String clientGuid) {

    boolean result = false;

    Stat stat = new Stat();

    byte[] data = getZooKeeper().getData(guidNodeName, false, stat);

    if (data != null && clientGuid.equals(new String(data))) {

      getZooKeeper().delete(guidNodeName, stat.getVersion());

      result = true;

    }

    return result;

  }

 

  /** 锁是否已经存在 */

  public boolean exists(String guidNodeName) {

    boolean result = false;

    Stat stat = getZooKeeper().exists(guidNodeName, false);

    result = stat != null;

    return result;

  }

}

3.3 基于子节点实现的分布式锁基类

ChildrenNodeLock.java

public abstract class ChildrenNodeLock extends ZooKeeperBase implements ZooKeeperLock {

  /** 获取当前节点的前一个节点，如果为空表示自己是第一个 */

  protected String getPrevElementName() {

    List<String> elementNames = getZooKeeper().getChildren(this.guidNodeName, false);

    long curElementSerial = Long.valueOf(

        elementNodeFullName.substring((this.guidNodeName + "/" + childPrefix).length()));

    String prevElementName = null;

    long prevElementSerial = -1;

    for (String oneElementName : elementNames) {

      long oneElementSerial = Long.parseLong(oneElementName.substring(childPrefix.length()));

      if (oneElementSerial < curElementSerial) {

        // 比当前节点小

        if (oneElementSerial > prevElementSerial) {

          prevElementSerial = oneElementSerial;

          prevElementName = oneElementName;

        }

      }

    }

    return prevElementName;

  }

 

  /** 尝试获取锁 */

  public boolean lock(String guidNodeName, String clientGuid) {

    boolean result = false;

    // 确保根节点存在，并且创建为容器节点

    super.createRootNode(this.guidNodeName, CreateMode.CONTAINER);

    // 创建子节点并返回带序列号的节点名

    elementNodeFullName = getZooKeeper().create(this.guidNodeName + "/" + childPrefix,

        new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

    boolean lockSuccess = isLockSuccess();

    result = lockSuccess;

    return result;

  }

 

 

  /** 释放锁 */

  public boolean release(String guidNodeName, String clientGuid) {

    // 删除子节点

    getZooKeeper().delete(elementNodeFullName, 0);

    return true;

  }

 

  /** 锁是否已经存在，容器节点存在，并且有子节点，则说明锁已经存在 */

  public boolean exists(String guidNodeName) {

    boolean exists = false;

    Stat stat = new Stat();

    try {

      getZooKeeper().getData(guidNodeName, false, stat);

      exists = stat.getNumChildren() > 0;

    } catch (KeeperException.NoNodeException e) {

      exists = false;

    }

    return exists;

  }

 

  /** 是否加锁成功 , 由子类实现 */

  protected abstract boolean isLockSuccess();

}

3.4 基于子节点实现的非阻塞锁

ChildrenBlocklessLock.java

public class ChildrenBlocklessLock extends ChildrenNodeLock {

  /** 是否加锁成功 */

  protected boolean isLockSuccess() throws KeeperException, InterruptedException {

    boolean lockSuccess = false;

    String prevElementName = getPrevElementName();

    if (prevElementName != null) {

      // 有更小的节点，说明当前节点没抢到锁，删掉自己并退出

      getZooKeeper().delete(elementNodeFullName, 0);

    } else {

      lockSuccess = true;

    }

    return lockSuccess;

  }

}

3.5 基于子节点实现的阻塞锁

ChildrenBlockingLock.java

public class ChildrenBlockingLock extends ChildrenNodeLock {

  /** 前一个节点被删除的信号 */

  static Integer mutex = Integer.valueOf(-1);

 

  /** 监控的节点被删除 */

  protected void processNodeDeleted(WatchedEvent event) {

    synchronized (mutex) {

      // 节点被删除，通知退出线程

      mutex.notify();

    }

  }

 

  /** 是否加锁成功 */

  protected boolean isLockSuccess() {

    boolean lockSuccess;

    while (true) {

      String prevElementName = getPrevElementName();

      if (prevElementName == null) {

        lockSuccess = true;

        break;

      } else {

        // 有更小的节点，说明当前节点没抢到锁，注册前一个节点的监听

        getZooKeeper().exists(this.guidNodeName + "/" + prevElementName, true);

        synchronized (mutex) {

          mutex.wait();

          log.info("{} 被删除，看看是不是轮到自己了", prevElementName);

        }

      }

    }

    return lockSuccess;

  }

}

4 测试用例

4.1 测试代码

LockClientThread.java 获取分布式锁和释放锁

public class LockClientThread extends Thread {

  /** 模拟获取分布式锁，成功后执行业务 */

  public void run() {

    boolean locked = zooKeeperLock.lock(guidNodeName, clientGuid);

    if (locked) {

      log.info("{} lock() success，拿到锁了，假装忙2秒", clientGuid);

      Thread.sleep(2000);

      boolean released = zooKeeperLock.release(guidNodeName, clientGuid);

      log.info("{} release() result ： {}", clientGuid, released);

    } else {

      log.info("{} lock() fail", clientGuid);

    }

  }

}

模拟多个客户端并发执行

public void testChildrenBlocklessMultiThread() throws IOException {

  String guidNodeName = "/multi-" + System.currentTimeMillis();

  int threadCount = 5;

 

  LockClientThread[] threads = new LockClientThread[threadCount];

  for (int i = 0; i < threadCount; i++) {

    ChildrenBlocklessLock nodeBlocklessLock = new ChildrenBlocklessLock(address);

    threads[i] = new LockClientThread(nodeBlocklessLock, guidNodeName, "client-" + (i + 1));

  }

  for (int i = 0; i < threadCount; i++) {

    threads[i].start();

  }

}

可以看到，只有一个线程能抢到锁并执行业务，其他线程都直接退出。

4.2 非阻塞锁的测试结果

55:43.929 [INFO] LockClientThread.run(33) client-1 lock() ...

55:43.942 [INFO] LockClientThread.run(33) client-3 lock() ...

55:43.947 [INFO] LockClientThread.run(33) client-2 lock() ...

55:43.948 [INFO] LockClientThread.run(33) client-4 lock() ...

55:43.949 [INFO] LockClientThread.run(33) client-5 lock() ...

55:44.052 [INFO] LockClientThread.run(36) client-1 lock() success，拿到锁了，假装忙2秒

55:44.072 [INFO] LockClientThread.run(47) client-5 lock() fail

55:44.085 [INFO] LockClientThread.run(47) client-4 lock() fail

55:44.091 [INFO] LockClientThread.run(47) client-2 lock() fail

55:44.096 [INFO] LockClientThread.run(47) client-3 lock() fail

55:46.053 [INFO] LockClientThread.run(42) client-1 release() ...

55:46.057 [INFO] LockClientThread.run(44) client-1 release() result ： true

4.3 阻塞锁的测试结果

可以看到，抢到分布式锁的线程执行业务，没抢到锁的线程会等到直到锁被释放重新获取到锁后再执行业务。

59:32.802 [INFO] LockClientThread.run(33) client-1 lock() ...

59:32.811 [INFO] LockClientThread.run(33) client-3 lock() ...

59:32.812 [INFO] LockClientThread.run(33) client-4 lock() ...

59:32.813 [INFO] LockClientThread.run(33) client-2 lock() ...

59:32.813 [INFO] LockClientThread.run(33) client-5 lock() ...

59:32.836 [INFO] LockClientThread.run(36) client-1 lock() success，拿到锁了，假装忙2秒

59:34.836 [INFO] LockClientThread.run(42) client-1 release() ...

59:34.844 [INFO] LockClientThread.run(44) client-1 release() result ： true

59:34.846 [INFO] ChildrenBlockingLock.isLockSuccess(55) element0000000000 被删除，看看是不是轮到自己了

59:34.848 [INFO] LockClientThread.run(36) client-5 lock() success，拿到锁了，假装忙2秒

59:36.848 [INFO] LockClientThread.run(42) client-5 release() ...

59:36.852 [INFO] ChildrenBlockingLock.isLockSuccess(55) element0000000001 被删除，看看是不是轮到自己了

59:36.852 [INFO] LockClientThread.run(44) client-5 release() result ： true

59:36.855 [INFO] LockClientThread.run(36) client-2 lock() success，拿到锁了，假装忙2秒

59:38.855 [INFO] LockClientThread.run(42) client-2 release() ...

59:38.869 [INFO] ChildrenBlockingLock.isLockSuccess(55) element0000000002 被删除，看看是不是轮到自己了

59:38.870 [INFO] LockClientThread.run(44) client-2 release() result ： true

59:38.876 [INFO] LockClientThread.run(36) client-4 lock() success，拿到锁了，假装忙2秒

59:40.877 [INFO] LockClientThread.run(42) client-4 release() ...

59:40.881 [INFO] ChildrenBlockingLock.isLockSuccess(55) element0000000003 被删除，看看是不是轮到自己了

59:40.882 [INFO] LockClientThread.run(44) client-4 release() result ： true

59:40.884 [INFO] LockClientThread.run(36) client-3 lock() success，拿到锁了，假装忙2秒

59:42.884 [INFO] LockClientThread.run(42) client-3 release() ...

59:42.887 [INFO] LockClientThread.run(44) client-3 release() result ： true