4.2 分布式锁

4.2 分布式锁

4.2 分布式锁

在分布式系统中，数据一致性是一个至关重要的问题。当多个服务节点需要并发访问共享资源时，如何保证数据的一致性和正确性，分布式锁就成为了解决这一问题的关键技术。本章节将深入探讨基于 Zookeeper 实现分布式锁的原理、实践以及相关代码详解。

4.2.1 分布式锁概述

什么是分布式锁？

分布式锁是控制分布式系统之间同步访问共享资源的一种方式。在单机环境中，我们通常使用线程锁（如 Java 中的 synchronized 或 ReentrantLock）来控制多线程对共享资源的并发访问。然而，在分布式环境中，由于资源分布在不同的机器上，传统的线程锁无法跨进程甚至跨机器工作，因此需要引入分布式锁。

为什么需要分布式锁？

考虑一个典型的电商场景：商品库存管理。假设多个服务器节点同时接收到用户下单请求，都需要更新商品库存。如果没有分布式锁的协调，可能会出现以下问题：

• 超卖问题: 多个节点同时读取到相同的库存数量，并进行减库存操作，导致实际售出的商品数量超过库存总量。

• 数据不一致: 不同节点更新库存的顺序不一致，导致最终库存数据混乱。

分布式锁的核心目标是保证在分布式环境下，同一时刻只有一个客户端能够获得锁，从而独占共享资源，防止数据冲突和不一致性问题。

分布式锁应具备的特性：

一个可靠的分布式锁应该具备以下特性：

• 互斥性 (Mutual Exclusion): 在任何时刻，只有一个客户端可以持有锁。这是分布式锁最基本也是最重要的特性。

• 避免死锁 (Deadlock-free): 即使持有锁的客户端发生故障，锁也能够在合理的时间内被释放，避免其他客户端永久等待。

• 容错性 (Fault Tolerance): 锁服务本身应该是高可用的，即使部分节点宕机，锁服务仍然能够正常运行。

• 可重入性 (Reentrancy) (可选): 同一个客户端在持有锁的情况下，可以再次请求获得该锁，避免自己阻塞自己。

• 公平性 (Fairness) (可选): 锁的获取应该是公平的，例如按照请求的先后顺序来分配锁，避免某些客户端长时间无法获得锁（饥饿问题）。

分布式锁的实现方式：

实现分布式锁的常见方式有很多，例如：

• 基于数据库: 利用数据库的唯一索引或排他锁来实现。

• 基于缓存 (Redis, Memcached): 利用缓存的原子操作 (如 SETNX) 和过期时间来实现。

• 基于分布式协调服务 (Zookeeper, etcd, Consul): 利用分布式协调服务的强一致性、临时节点和 Watcher 机制来实现。

本章节重点探讨基于 Zookeeper 实现分布式锁。

4.2.2 基于 Zookeeper 实现分布式锁的原理

Zookeeper 是一个高性能的分布式协调服务，其核心特性包括：

• 数据模型: 类似于文件系统的树形结构，节点称为 ZNode。

• 临时节点 (Ephemeral Node): 客户端与 Zookeeper 服务器断开连接后，临时节点会被自动删除。

• 顺序节点 (Sequential Node): 创建节点时，Zookeeper 会自动在节点名称后追加一个单调递增的数字。

• Watcher 机制: 客户端可以注册 Watcher 监听 ZNode 的变化，当 ZNode 发生变化时，Zookeeper 会通知客户端。

• 强一致性 (Strong Consistency): Zookeeper 保证数据在所有服务器节点之间的一致性。

基于 Zookeeper 实现分布式锁，主要是利用其临时顺序节点和 Watcher 机制。

独占锁 (Exclusive Lock) 实现原理：

独占锁是最常见的分布式锁类型，也称为写锁。它保证在任何时刻，只有一个客户端可以持有锁，用于保护共享资源的写操作。

加锁流程：

1. 创建临时顺序节点: 客户端在指定的 Zookeeper 节点下创建一个临时顺序节点，例如 /locks/lock-0000000001，/locks/lock-0000000002，等等。

2. 获取子节点列表: 客户端获取 /locks 节点下的所有子节点列表。

3. 判断是否为最小节点: 客户端判断自己创建的节点是否是所有子节点中序号最小的节点。

   • 如果是最小节点: 则客户端获得锁。

   • 如果不是最小节点: 则客户端需要监听比自己序号小的那个节点的删除事件 (Watcher)。

4. 监听前一个节点的删除事件: 如果客户端不是最小节点，则监听序号仅次于自己创建的节点的前一个节点的删除事件。例如，如果客户端创建的节点是 /locks/lock-0000000003，则监听 /locks/lock-0000000002 的删除事件。

5. 收到删除事件后重新尝试获取锁: 当监听的前一个节点被删除后，客户端会收到 Watcher 通知，此时客户端需要重新执行步骤 2 和 3，再次尝试获取锁。

释放锁流程：

当客户端完成共享资源的访问后，需要释放锁。释放锁的过程非常简单，只需要删除自己创建的临时顺序节点即可。由于是临时节点，即使客户端发生故障，连接断开，Zookeeper 也会自动删除该节点，从而避免死锁。

流程图 (graph TD):

独占锁的优势：

• 可靠性高: 基于 Zookeeper 实现，利用其强一致性和可靠性，锁服务本身具有高可用性和容错性。

• 避免死锁: 临时节点机制保证了即使客户端崩溃，锁也会被自动释放。

• 实现简单: 原理相对简单，易于理解和实现。

独占锁的缺点：

• 性能略低: 每次加锁和释放锁都需要与 Zookeeper 服务器进行多次网络通信，性能相比于本地锁略低。

• 羊群效应 (Herd Effect): 当锁被释放时，所有监听该锁的客户端都会收到通知并尝试获取锁，可能会造成瞬间的性能冲击。

公平锁和非公平锁：

默认情况下，上述实现的独占锁是非公平锁。当锁被释放时，所有等待的客户端都会尝试抢占锁，没有明确的顺序保证。

如果需要实现公平锁，即按照请求的先后顺序来获取锁，可以稍作修改：

• 公平锁实现: 在判断是否为最小节点时，不仅要判断自己是否是最小节点，还要判断自己是否是所有等待节点中最早请求锁的节点。可以通过记录每个客户端请求锁的时间戳来实现。

可重入锁 (Reentrant Lock) 实现：

上述实现的独占锁是非可重入锁。如果同一个客户端在持有锁的情况下再次请求获取锁，会被阻塞。

如果需要实现可重入锁，即允许同一个客户端多次获取同一个锁，可以进行如下改进：

• 记录锁持有者: 在创建临时顺序节点时，将客户端的标识信息 (例如客户端 IP 地址或 Session ID) 写入节点的数据中。

• 判断锁持有者: 在判断是否获得锁时，如果发现自己创建的节点不是最小节点，但前一个节点的持有者是自己，则也认为获得了锁 (重入)。

• 计数器: 在节点数据中维护一个计数器，记录客户端重入锁的次数。每次重入计数器加一，每次释放锁计数器减一，当计数器为零时才真正删除节点。

4.2.3 分布式锁代码实践 (Java + Zookeeper)

下面提供一个基于 Java 和 Zookeeper 实现独占锁的代码示例。

依赖引入 (Maven):

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.6.3</version> <!-- 使用最新稳定版本 -->
</dependency>
​

Zookeeper 分布式锁类 (ZookeeperDistributedLock.java):

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
public class ZookeeperDistributedLock {
    private ZooKeeper zk;
    private String lockBasePath;
    private String lockNodeName;
    private String currentLockPath;
    private CountDownLatch latch = new CountDownLatch(1);
    public ZookeeperDistributedLock(String zkConnectionString, String lockBasePath, String lockNodeName) throws IOException, InterruptedException, KeeperException {
        this.lockBasePath = lockBasePath;
        this.lockNodeName = lockNodeName;
        this.zk = new ZooKeeper(zkConnectionString, 3000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    latch.countDown(); // 连接成功，CountDownLatch 计数减一
                }
            }
        });
        latch.await(); // 等待 Zookeeper 连接成功
        Stat stat = zk.exists(lockBasePath, false);
        if (stat == null) {
            zk.create(lockBasePath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); // 创建锁根节点
        }
    }
    // 获取锁
    public boolean acquireLock(long timeout) throws KeeperException, InterruptedException {
        try {
            currentLockPath = zk.create(lockBasePath + "/" + lockNodeName + "-", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            return tryLock(timeout);
        } catch (KeeperException e) {
            deleteCurrentLock(); // 创建节点失败，清理
            throw e;
        } catch (InterruptedException e) {
            deleteCurrentLock(); // 创建节点失败，清理
            throw e;
        }
    }
    // 尝试获取锁
    private boolean tryLock(long timeout) throws KeeperException, InterruptedException {
        long startTime = System.currentTimeMillis();
        while (true) {
            List<String> childrenNodes = zk.getChildren(lockBasePath, false);
            Collections.sort(childrenNodes); // 排序子节点
            String smallestNode = childrenNodes.get(0);
            if (currentLockPath.endsWith(smallestNode)) {
                return true; // 当前节点是最小节点，获得锁
            }
            String predecessorNode = null;
            int currentNodeIndex = childrenNodes.indexOf(currentLockPath.substring(lockBasePath.length() + 1));
            if (currentNodeIndex > 0) {
                predecessorNode = childrenNodes.get(currentNodeIndex - 1);
            }
            if (predecessorNode != null) {
                Stat stat = zk.exists(lockBasePath + "/" + predecessorNode, new Watcher() { // 监听前一个节点删除事件
                    @Override
                    public void process(WatchedEvent event) {
                        if (event.getType() == Event.EventType.NodeDeleted) {
                            latch.countDown(); // 前一个节点删除，CountDownLatch 计数减一，唤醒等待线程
                        }
                    }
                });
                if (stat != null) {
                    latch = new CountDownLatch(1); // 重置 CountDownLatch
                    long remainingTime = timeout - (System.currentTimeMillis() - startTime);
                    if (remainingTime <= 0) {
                        deleteCurrentLock(); // 超时未获得锁，释放当前创建的节点
                        return false; // 获取锁超时
                    }
                    latch.await(remainingTime, TimeUnit.MILLISECONDS); // 等待前一个节点释放锁或超时
                } else {
                    // 前一个节点不存在，可能已经被删除，重新尝试获取锁
                }
            } else {
                // 理论上不应该发生，除非根节点下没有子节点了，需要重新获取子节点列表
            }
            if (timeout > 0 && (System.currentTimeMillis() - startTime) > timeout) {
                deleteCurrentLock(); // 超时未获得锁，释放当前创建的节点
                return false; // 获取锁超时
            }
        }
    }
    // 释放锁
    public void releaseLock() throws KeeperException, InterruptedException {
        deleteCurrentLock();
    }
    // 删除当前锁节点
    private void deleteCurrentLock() throws KeeperException, InterruptedException {
        if (currentLockPath != null) {
            zk.delete(currentLockPath, -1);
            currentLockPath = null;
        }
    }
    // 关闭 Zookeeper 连接
    public void close() throws InterruptedException {
        if (zk != null) {
            zk.close();
        }
    }
}
​

代码详解：

• 构造函数:

  • 接收 Zookeeper 连接字符串 (zkConnectionString)、锁的根路径 (lockBasePath) 和锁节点名称 (lockNodeName) 作为参数。

  • 创建 ZooKeeper 客户端实例，并使用 CountDownLatch 保证 Zookeeper 连接成功后才继续执行。

  • 检查锁的根节点是否存在，如果不存在则创建持久节点作为根节点。

• acquireLock(long timeout) 方法:

  • 创建临时顺序节点，节点名称格式为 lockBasePath + "/" + lockNodeName + "-" + 序号。例如 /locks/my-lock-0000000001。

  • 调用 tryLock(timeout) 方法尝试获取锁。

• tryLock(long timeout) 方法:

  • 循环尝试获取锁，直到获得锁或超时。

  • 获取锁根节点下的所有子节点，并排序。

  • 判断当前客户端创建的节点是否是序号最小的节点。如果是，则获得锁，返回 true。

  • 如果不是最小节点，则获取序号仅次于当前节点的前一个节点。

  • 监听前一个节点的删除事件，使用 CountDownLatch 实现等待机制。当收到前一个节点删除事件时，CountDownLatch 计数减一，唤醒等待线程，重新尝试获取锁。

  • 如果超时时间到，仍然没有获得锁，则删除当前创建的节点，返回 false。

• releaseLock() 方法:

  • 调用 deleteCurrentLock() 方法释放锁。

• deleteCurrentLock() 方法:

  • 删除当前客户端创建的临时顺序节点。

• close() 方法:

  • 关闭 Zookeeper 连接。

使用示例 (Main.java):

import org.apache.zookeeper.KeeperException;
import java.io.IOException;
public class Main {
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        String zkConnectionString = "localhost:2181"; // Zookeeper 连接字符串
        String lockBasePath = "/locks"; // 锁根路径
        String lockName = "my-lock"; // 锁名称
        ZookeeperDistributedLock lock = new ZookeeperDistributedLock(zkConnectionString, lockBasePath, lockName);
        try {
            System.out.println("尝试获取锁...");
            if (lock.acquireLock(5000)) { // 尝试获取锁，超时时间 5 秒
                System.out.println("成功获取锁！");
                // 模拟共享资源访问
                System.out.println("访问共享资源...");
                Thread.sleep(3000); // 模拟业务处理
                System.out.println("共享资源访问完成，释放锁...");
                lock.releaseLock(); // 释放锁
                System.out.println("锁已释放。");
            } else {
                System.out.println("获取锁超时！");
            }
        } finally {
            lock.close(); // 关闭 Zookeeper 连接
        }
    }
}
​

运行步骤：

1. 确保本地已启动 Zookeeper 服务 (例如使用 docker run --name zk -d -p 2181:2181 zookeeper:latest)。

2. 编译并运行 Main.java 程序。

3. 可以多次运行 Main.java 程序，模拟多个客户端竞争锁的场景。

代码说明：

• 代码示例实现了一个基本的独占锁，具备互斥性和避免死锁的特性。

• 代码中使用了 CountDownLatch 来实现线程等待和唤醒机制，简化了 Watcher 的使用。

• 代码中加入了超时机制，防止客户端长时间阻塞等待锁。

• 实际应用中，需要根据业务场景进行更完善的错误处理、重试机制和日志记录。

• 可以扩展代码，实现公平锁、可重入锁等更高级的分布式锁功能。

4.2.4 分布式锁的适用场景和注意事项

适用场景：

• 资源竞争: 多个分布式节点需要并发访问共享资源，例如数据库记录、文件、队列等。

• 任务调度: 在分布式任务调度系统中，需要保证同一时刻只有一个节点能够执行某个任务。

• 领导者选举: 在分布式系统中，可以使用分布式锁来实现领导者选举，保证只有一个节点成为领导者。

• 配置中心: 在分布式配置中心中，可以使用分布式锁来保证配置更新的原子性。

注意事项：

• 锁的粒度: 选择合适的锁粒度非常重要。锁粒度过粗会导致并发度降低，锁粒度过细会增加锁管理的复杂性。

• 锁的超时时间: 设置合理的锁超时时间，避免客户端长时间持有锁不释放，影响系统性能。

• 网络延迟和抖动: 分布式锁的性能会受到网络延迟和抖动的影响，需要根据实际网络环境进行性能评估和优化。

• Zookeeper 集群稳定性: Zookeeper 集群的稳定性直接影响分布式锁的可靠性，需要保证 Zookeeper 集群的稳定运行。

• 避免过度依赖分布式锁: 分布式锁虽然重要，但不是万能的。应该尽量减少对分布式锁的依赖，通过优化业务逻辑和数据结构来减少并发冲突。

4.2.5 总结

本章节详细介绍了基于 Zookeeper 实现分布式锁的原理、代码实践和应用场景。Zookeeper 分布式锁利用其临时顺序节点和 Watcher 机制，能够有效地解决分布式系统中的资源并发访问问题，保证数据的一致性和正确性。

核心要点回顾：

• 分布式锁是解决分布式系统并发控制的关键技术。

• Zookeeper 基于临时顺序节点和 Watcher 机制实现分布式锁。

• 独占锁是最常见的分布式锁类型，保证同一时刻只有一个客户端持有锁。

• 基于 Zookeeper 实现分布式锁具有可靠性高、避免死锁等优点，但也存在性能略低、羊群效应等缺点。

• 需要根据实际业务场景选择合适的分布式锁实现方式，并注意锁的粒度、超时时间、网络延迟等问题。

希望本章节能够帮助读者深入理解 Zookeeper 分布式锁的原理和实践，并在实际项目中灵活应用。