分布式锁实战-基于etcd的实现很优雅

篇幅太长看着也累，每天进步一点点

欢迎关注公众号「架构染色」交流和学习

1. etcd 简介

1.1 etcd 的背景

虽然Kubernetes 给云原生时代带来了颠覆性的新气象，但却很少人了解被钦定作为其后端存储的 etcd ，本篇从分布式锁视角梳理etcd的各种机制，探索基于etcd的锁实现是怎样。

etcd 能被Kubernetes 如此青睐，是因为它一直在聆听社区的声音并快速改进，积极配合 Kubernetes 项目向前推进，解决社区反馈的痛点；发起 V2 到 V3 的重大版本更新，尤其是 19 年由 Google、Alibaba 等公司联合打造的 3.4 版本，满足了 Kubernetes 在超大型公司大规模使用中严苛的可用性、扩展性和性能等要求。

上图描述了 etcd 名字的由来，其寓意是为大规模分布式系统提供存储配置信息；华为 Kubernetes 专家杜军老师专门为 etcd 著书《云原生分布式存储基石-etcd》，称它基石足见何等重要，另一位大咖etcd 的作者李响老师曾讲：“etcd 就是用来存储云上最重要的数据的”；相信随着云原生架构的演进，etcd 会担任越来越多重要的角色；技术人员在云原生时代，除了拥抱 Kubernetes，也要拥抱 etcd。

1.2 etcd vs ZooKeeper

很长一段时间 ZooKeeper（后文简称 ZK）被作为默认首选项，用于解决分布式系统的协同和元数据存储，但其太复杂、迭代慢、难维护、性能缺陷等问题逐渐成为槽点；而 etcd 吸取了 ZK 的教训，从设计和实现上具有后见之明，提供了更好的工程和运维体验，其主要改进在于如下几个方面：

动态的集群节点关系重配置
高负载条件下的稳定读写
多版本并发控制的数据模型
持久、稳定的监听机制
租约 (lease) 原语实现了连接与会话的解耦
安全的分布式共享锁 API
普适的 HTTP 、GRPC 通信协议

据说在一个由 3 台 8 核节点组成的云服务器上, etcd v3 版本可以做到每秒数万次的写操作和数十万次的读操作（ZK：）；后续会有其他篇章结合QA的测试结果来探讨etcd服务端的设计和性能情况。

1.3 etcd 特性介绍

为满足本篇目标所需，也考虑到对 etcd 熟悉的读者不多，这里着重介绍以下几个关键特性：

数据组织：etcd 在 V2 版本时跟 ZK 一样以树形目录结构来组织数据，V3 版本则通过半开区间（ Key range）取代 V2 中的目录结构，优化成扁平的 Key-Value 结构，Key 仍采用之前的格式如/lock/lock1/uuid1、/lock/lock1/uuid2，感官跟树形目录形式保持一致，但实际是一个完整独立的 key；可以基于前缀（Prefix）机制通过/lock/lock1/查询一批拥有相同前缀的数据，但其语义是前缀匹配查询而非目录子节点查询；这个改造也是为了支撑更多的特性和达到更好的性能
集群模式：通常是由 3、5 个基数实例组成集群，当超过半数服务实例正常工作就能对外提供服务，既能避免单点故障，又尽量高可用，每个服务实例都有一个数据备份，通过 raft 协议实现数据全局一致
顺序更新：每个 etcd 节点都可接收读写请求，但变更类（增删改）请求会在集群内转给 leader 执行，来所有客户端的变更请求将按照 leader 接收的顺序被处理，在 ZK 中插入同名节点，ZK 会自动为同名节点名后添加递增序号，即避免冲突又具有顺序管控的意义；但 etcd 中不能插入同名 Key，它是采用 revision 机制来管控更新类请求的顺序，使用一个全局计数器，单调递增，每当有数据变更，revision 就会加一（具有全局唯一性），而每个 revision 都关联对应修改的数据，可以通过 revision 的大小推断数据变更的顺序，利用这个特性可以实现高级协调服务
租约（Lease）机制：常见的 TTL（Time To Live）机制是给单个 Key-Value 设置存活时间，超过时间后自动删除这个 Key-Value，V3 中加入的租约机制更高级一些，针对每个 Lease 设置了一个 TTL 时间，存储 Key-Value 时可指定一个 Lease，多个拥有相同 TTL 的 key 绑定到同一个 Lease，实现 Lease 的复用。同时也支持续约，可以通过客户端在 Lease 到期之前续约，以避免 Key-Value 过期后失效，也可以通过客户端主动解约，这比 ZK 中基于 session 的状态保持更灵活
监听机制：客户端可以注册对某一个 Key 或一批相同前缀（前缀机制，似 ZK 中的父节点）Key 的监听，当被监听的这些 Key 发生变更，客户端将收到通知，感知到变更。

etcd 的分布式锁正是基于以上特性来实现的，简单来说是：

租约机制：用于支撑异常情况下的锁自动释放能力
前缀和 Revision 机制：用于支撑公平获取锁和排队等待的能力
监听机制：用于支撑抢锁能力
集群模式：用于支撑锁服务的高可用

2. 加解锁的流程描述

准备客户端和 Key

客户端连接 Etcd，获取 LockClient，获取 LeaseClient
以/lock/lock1 为前缀 + / + uuid 创建全局唯一的 key，如：

client-a 线程 1 的 key 为”/lock/lock1/uuid1″，申请 lock1
client-a 线程 2 的 key 为”/lock/lock1/uuid2″，申请 lock1
client-c 线程 1 的 key 为”/lock/lock2/uuid3″，申请 lock2
client-b 线程 1 的 key 为”/lock/lock2/uuid4″，申请 lock2

客户端 a、b、c 分别创建租约，租约的时长由业务输入确定

创建租约并保持续租

方案 1-创建定时任务定时续租，无论客户端是持锁状态还是等待锁状态 Key 都必须存在，而 Key 是否释放由租约管控，所以都需要保持租约的活性

持锁状态：当业务未完成时，不能让租约到期，需定时续租；当业务完成时可主动解除租约，持锁 Key 会被删除；若客户端异常，租约到期后持锁 Key 也会被删除；等锁的客户端监听到持锁 Key 被删除后，可开始抢锁。
等锁状态：等锁超时会主动解除租约，或客户端异常时等锁 key 被删除，后边排队的就前进一步，尝试抢锁。

方案 2-使用自动续约

绑定租约写 key

每个客户端在执行 put 操作时，将第 1 步中准备的具有唯一性的 Key 绑定租约写入 etcd

获取竞争锁的 key-Value 列表

以 lock1 为例，客户端以前缀”/lock/lock1″ 读取所有匹配此前缀的的 key-Value 列表（key-Value 中带有 key 对应的 Revision）

对所获取的 Key-Value 列表按 revision 从小到大排序
判断自己是不是第一个（revision 最小），若是，则成功获取锁
若不是，则监听自己的前一个 Key-Value 的删除事件

删除事件由主动释放租约或者因租约过期失效而触发
这种只监听前一个 Key-Value 的方式避免了惊群效应问题

若是阻塞申请锁，则申请锁的操作可增加阻塞等待
若监听事件生效，则回到第 4 步重新进行判断，直到获取到锁
解锁时，将第一个 Key-Value 的租约释放

3. etcd 分布式锁的能力

可能读者是单篇阅读，这里引入第一篇《分布式锁上-初探》中的一些内容，一个分布式锁应具备这样一些功能特点：

互斥性：在同一时刻，只有一个客户端能持有锁
安全性：避免死锁，如果某个客户端获得锁之后处理时间超过最大约定时间，或者持锁期间发生了故障导致无法主动释放锁，其持有的锁也能够被其他机制正确释放，并保证后续其它客户端也能加锁，整个处理流程继续正常执行
可用性：也被称作容错性，分布式锁需要有高可用能力，避免单点故障，当提供锁的服务节点故障（宕机）时不影响服务运行，这里有两种模式：一种是分布式锁服务自身具备集群模式，遇到故障能自动切换恢复工作；另一种是客户端向多个独立的锁服务发起请求，当某个锁服务故障时仍然可以从其他锁服务读取到锁信息(Redlock)
可重入性：对同一个锁，加锁和解锁必须是同一个线程，即不能把其他线程持有的锁给释放了
高效灵活：加锁、解锁的速度要快；支持阻塞和非阻塞；支持公平锁和非公平锁

基于上边对 etcd 分布式锁的介绍，这里简单总结一下 etcd 的能力矩阵，ZK 的情况请看《分布式锁中-基于 Zookeeper 的实现》，redis锁的情况会在后续文章中补充

能力	ZK	etcd
互斥	是	是
安全	链接异常时，session 丢失自动释放锁	基于租约，租约过期后自动释放锁，不用像ZK那样释放链接
可用性	相对可用性还好	好
可重入	服务端非可重入，本地线程可重入	服务端非可重入，本地线程可重入需自研
加解锁速度	速度不算快	速度快，如GRPC 协议优势、服务端性能的优势
阻塞非阻塞	客户端两种能力都提供	jetcd-core 中，阻塞非阻塞由 Future#get 的超时控制能力支撑
公平非公平	公平锁	公平锁
可续租	天然支持，基于session	天然支持，基于Lease

4. jetcd 库实现分布式锁

etcd 针对 java 语言的客户端有官方的 jetcd-core 还有 IBM 的 etcd-java，下文使用 jetcd-core 做示例。

jetcd-core 中提供了高阶的 Lock API（无需使用者准备唯一 key、前缀查询 Key-Value 列表、排序判断自己是否 revision 最小的、监听前一个 Key-Value 的删除），使用者只需关注最原始的诉求：申请的锁是什么名称、用多久，申请不到就等多久；使用高阶 API 实现分布式锁的流程会比第 2 部分原生的流程要简单许多，流程如下：

连接 etcd 集群
获取 LockClient，获取 LeaseClient，不需要准备 key
创建租约并保持续租，你觉得哪一种方案会更好呢

方案 1-创建定时任务定时续租
方案 2-使用自动续约

绑定租约抢锁
若抢到锁，则返回锁路径信息
若没抢到锁，则阻塞等待（可设置超时返回）
解锁时，传入锁信息解锁；根据业务场景决定如何解约

4.1 pom 依赖

<dependency>
  <groupId>io.etcd</groupId>
  <artifactId>jetcd-core</artifactId>
  <version>0.7.3</version>
</dependency>

4.2 Lease 相关的 API 介绍

public interface Lease extends CloseableClient {

     //创建一个租约，过期时间是ttl,单位秒；没有请求超时控制
    CompletableFuture<LeaseGrantResponse> grant(long ttl);

     //创建一个租约，过期时间是ttl,单位秒；后两个参数是请求超时控制
    CompletableFuture<LeaseGrantResponse> grant(long ttl, long timeout, TimeUnit unit);

     //解约
    CompletableFuture<LeaseRevokeResponse> revoke(long leaseId);

  //主动续约1次
    CompletableFuture<LeaseKeepAliveResponse> keepAliveOnce(long leaseId);

     //查询租约信息
    CompletableFuture<LeaseTimeToLiveResponse> timeToLive(long leaseId, LeaseOption leaseOption);

     //自动续约
    CloseableClient keepAlive(long leaseId, StreamObserver<LeaseKeepAliveResponse> observer);
}

4.3 lock 相关的 API 介绍

public interface Lock extends CloseableClient {

    //绑定租约申请指定名称的锁，抢锁成功返回锁秘钥，
    CompletableFuture<LockResponse> lock(ByteSequence name, long leaseId);

    //持锁秘钥解锁
    CompletableFuture<UnlockResponse> unlock(ByteSequence lockKey);
}

4.4 分布式锁示例

package com.rock.dlock;

import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.Lease;
import io.etcd.jetcd.Lock;
import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
import io.etcd.jetcd.lock.LockResponse;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @author zs
 * @date 2022/11/6 12:23 PM
 */
public class DemoEtcdLock {
    private final static Logger log = LoggerFactory.getLogger(DemoEtcdLock.class);
    private Client client;
    private Lock lockClient;
    private Lease leaseClient;

    private LockState lockState;

    class LockState{
        private String lockKey;
        private String lockPath;
        private String errorMsg;
        private long leaseTTL;
        private long leaseId;
        private boolean lockSuccess;

        public LockState(String lockKey, long leaseTTL) {
            this.lockKey = lockKey;
            this.leaseTTL = leaseTTL;
        }

        public String getLockKey() {
            return lockKey;
        }

        public void setLockKey(String lockKey) {
            this.lockKey = lockKey;
        }

        public String getLockPath() {
            return lockPath;
        }

        public void setLockPath(String lockPath) {
            this.lockPath = lockPath;
        }

        public String getErrorMsg() {
            return errorMsg;
        }

        public void setErrorMsg(String errorMsg) {
            this.errorMsg = errorMsg;
        }

        public long getLeaseId() {
            return leaseId;
        }

        public void setLeaseId(long leaseId) {
            this.leaseId = leaseId;
        }

        public boolean isLockSuccess() {
            return lockSuccess;
        }

        public void setLockSuccess(boolean lockSuccess) {
            this.lockSuccess = lockSuccess;
        }

        public long getLeaseTTL() {
            return leaseTTL;
        }

        public void setLeaseTTL(long leaseTTL) {
            this.leaseTTL = leaseTTL;
        }
    }


    public DemoEtcdLock(Client client, String lockKey, Long leaseTTL, TimeUnit unit) {
        this.client = client;
        //1.准备客户端
        this.lockClient = client.getLockClient();
        this.leaseClient = client.getLeaseClient();
        this.lockState = new LockState(lockKey,unit.toSeconds(leaseTTL));
    }


    public boolean lock()  {
        try {
            //2.创建租约,并自动续约
            createLease();

            //3.执行加锁，并为锁对应的Key绑定租约
            createLock();
        }catch (InterruptedException | ExecutionException e) {
            //todo:异常处理
        }
        return lockState.isLockSuccess();
    }

    public void unlock()  {
        try {
        //正常释放锁
            if (this.lockState.getLockPath() != null) {
                lockClient.unlock(ByteSequence.from(lockState.getLockPath().getBytes())).get();
            }
            //如果是主动续约，则关闭续约的定时任务

            //删除租约
            if (lockState.getLeaseId() != 0L) {
                leaseClient.revoke(lockState.getLeaseId());
            }
        } catch (InterruptedException | ExecutionException e) {
            //todo:异常处理
        }
        log.info("线程：{} 释放锁", Thread.currentThread().getName());
    }

    // 创建一个租约
    private void createLease() throws ExecutionException, InterruptedException {
        log.debug("[etcd-lock]: start to createLease." + this.lockState.getLockKey() + Thread.currentThread().getName());
        try {
            long leaseId = leaseClient.grant(this.lockState.getLeaseTTL()).get().getID();
            lockState.setLeaseId(leaseId);
            //自动续约
            StreamObserver<LeaseKeepAliveResponse> observer = new StreamObserver<LeaseKeepAliveResponse>() {
                @Override
                public void onNext(LeaseKeepAliveResponse value) {
                    log.trace("cluster node lease remaining ttl: {}, lease id: {}", value.getTTL(), value.getID());
                }

                @Override
                public void onError(Throwable t) {
                    log.error("cluster node lease keep alive failed. exception info: {}", t);
                }

                @Override
                public void onCompleted() {
                    log.trace("cluster node lease completed");
                }
            };
            // 设置自动续约
            leaseClient.keepAlive(leaseId, observer);
        }catch (InterruptedException | ExecutionException e) {
            log.error("[etcd-lock] Create lease failed:" + e);
            lockState.setErrorMsg("Create lease failed:" + e);
            throw e;
        }
    }

    private void createLock() throws ExecutionException, InterruptedException {
        String lockKey = this.lockState.getLockKey();
        log.debug("[etcd-lock]: start to createLock." + lockKey + Thread.currentThread().getName());
        try {
            LockResponse lockResponse = lockClient.lock(ByteSequence.from(lockKey.getBytes()), lockState.getLeaseId()).get();
            if (lockResponse != null) {
                String lockPath = lockResponse.getKey().toString(StandardCharsets.UTF_8);
                this.lockState.setLockPath(lockPath);
                log.info("线程：{} 加锁成功，锁路径：{}", Thread.currentThread().getName(), lockPath);
                this.lockState.setLockSuccess(true);
            }
        }
        catch (InterruptedException | ExecutionException e) {
            log.error("[etcd-lock] lock failed:" + e);
            lockState.setErrorMsg("[etcd-lock] lock failed:" + e);
            leaseClient.revoke(this.lockState.getLeaseId());
            throw e;
        }
    }
}

测试锁

package com.rock.dlock;

import io.etcd.jetcd.Client;

import java.util.concurrent.TimeUnit;

/**
 * @author zs
 * @date 2022/11/6 12:23 PM
 */
public class TestEtcdLock {
    public static void main(String[] args) {
        Client client = Client.builder().endpoints("http://localhost:2379").build();

        DemoEtcdLock demoEtcdLock1 = new DemoEtcdLock(client,"rock",30L, TimeUnit.SECONDS);
        DemoEtcdLock demoEtcdLock2 = new DemoEtcdLock(client,"rock",30L, TimeUnit.SECONDS);

        boolean lock1 = demoEtcdLock1.lock();
        if(lock1) {
            try {
                System.out.printf("do something");
            } finally {
                demoEtcdLock1.unlock();
            }
        }
        demoEtcdLock1.lock();//demoEtcdLock1 持锁未释放
        demoEtcdLock2.lock();//demoEtcdLock2 客户端无可重入设计，这里将会阻塞等待demoEtcdLock1释放锁
    }
}