Java如何实现ZooKeeper分布式锁

在分布式系统中，多个进程或线程可能需要同时访问共享资源。为了避免竞争条件和数据不一致的问题，分布式锁是一种常见的解决方案。ZooKeeper分布式协调服务，提供了实现分布式锁的能力。本文将介绍如何使用Java和ZooKeeper实现分布式锁。

1. ZooKeeper简介

ZooKeeper是一个开源的分布式协调服务，由Apache软件基金会维护。它提供了一个简单的接口，用于管理分布式系统中的配置信息、命名服务、分布式同步和组服务。ZooKeeper的核心是一个分布式的、层次化的文件系统，类似于文件系统的目录树结构。

2. 分布式锁的基本概念

分布式锁是一种用于在分布式系统中控制对共享资源访问的机制。它确保在同一时间只有一个进程或线程可以访问共享资源，从而避免竞争条件和数据不一致的问题。

ZooKeeper实现分布式锁的基本思路是利用ZooKeeper的临时顺序节点（Ephemeral Sequential Node）。每个客户端在尝试获取锁时，会在ZooKeeper中创建一个临时顺序节点。ZooKeeper会为每个节点分配一个唯一的顺序号。客户端通过比较自己创建的节点的顺序号与所有其他节点的顺序号，来判断自己是否获得了锁。

3. 实现步骤

3.1 创建ZooKeeper客户端

首先，我们需要创建一个ZooKeeper客户端，用于与ZooKeeper服务器进行通信。

import org.apache.zookeeper.ZooKeeper;

public class DistributedLock {
    private ZooKeeper zk;
    private String lockPath;

    public DistributedLock(String zkAddress, String lockPath) throws Exception {
        this.zk = new ZooKeeper(zkAddress, 3000, null);
        this.lockPath = lockPath;
    }
}

3.2 创建临时顺序节点

在尝试获取锁时，客户端需要在ZooKeeper中创建一个临时顺序节点。

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;

public class DistributedLock {
    // ... 其他代码

    public String createLockNode() throws Exception {
        return zk.create(lockPath + "/lock_", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    }
}

3.3 获取锁

客户端在创建节点后，需要检查自己创建的节点是否是最小的顺序节点。如果是，则表示获取锁成功；否则，客户端需要等待，直到前面的节点被释放。

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.util.List;
import java.util.concurrent.CountDownLatch;

public class DistributedLock {
    // ... 其他代码

    public boolean tryLock() throws Exception {
        String lockNode = createLockNode();
        while (true) {
            List<String> children = zk.getChildren(lockPath, false);
            String smallestNode = children.stream().min(String::compareTo).orElse(null);
            if (lockNode.endsWith(smallestNode)) {
                return true;
            } else {
                CountDownLatch latch = new CountDownLatch(1);
                Watcher watcher = new Watcher() {
                    @Override
                    public void process(WatchedEvent event) {
                        latch.countDown();
                    }
                };
                zk.getChildren(lockPath, watcher);
                latch.await();
            }
        }
    }
}

3.4 释放锁

当客户端完成对共享资源的访问后，需要释放锁。释放锁的操作是删除自己创建的临时顺序节点。

public class DistributedLock {
    // ... 其他代码

    public void releaseLock(String lockNode) throws Exception {
        zk.delete(lockNode, -1);
    }
}

4. 完整示例

以下是一个完整的示例，展示了如何使用ZooKeeper实现分布式锁。

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;

import java.util.List;
import java.util.concurrent.CountDownLatch;

public class DistributedLock {
    private ZooKeeper zk;
    private String lockPath;

    public DistributedLock(String zkAddress, String lockPath) throws Exception {
        this.zk = new ZooKeeper(zkAddress, 3000, null);
        this.lockPath = lockPath;
    }

    public String createLockNode() throws Exception {
        return zk.create(lockPath + "/lock_", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    }

    public boolean tryLock() throws Exception {
        String lockNode = createLockNode();
        while (true) {
            List<String> children = zk.getChildren(lockPath, false);
            String smallestNode = children.stream().min(String::compareTo).orElse(null);
            if (lockNode.endsWith(smallestNode)) {
                return true;
            } else {
                CountDownLatch latch = new CountDownLatch(1);
                Watcher watcher = new Watcher() {
                    @Override
                    public void process(WatchedEvent event) {
                        latch.countDown();
                    }
                };
                zk.getChildren(lockPath, watcher);
                latch.await();
            }
        }
    }

    public void releaseLock(String lockNode) throws Exception {
        zk.delete(lockNode, -1);
    }

    public static void main(String[] args) throws Exception {
        DistributedLock lock = new DistributedLock("localhost:2181", "/locks");
        String lockNode = null;
        try {
            if (lock.tryLock()) {
                lockNode = lock.createLockNode();
                System.out.println("Lock acquired");
                // 访问共享资源
            }
        } finally {
            if (lockNode != null) {
                lock.releaseLock(lockNode);
                System.out.println("Lock released");
            }
        }
    }
}

5. 总结

通过使用ZooKeeper的临时顺序节点，我们可以实现一个简单而有效的分布式锁。这种锁机制可以确保在分布式系统中，多个进程或线程能够安全地访问共享资源，避免竞争条件和数据不一致的问题。在实际应用中，还可以结合其他技术（如超时机制、重试机制等）来增强分布式锁的健壮性和可靠性。