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

public interface Lease extends CloseableClient {

     //創(chuàng)建一個租約，過期時間是ttl,單位秒；沒有請求超時控制
    CompletableFuture<LeaseGrantResponse> grant(long ttl);

     //創(chuàng)建一個租約，過期時間是ttl,單位秒；后兩個參數是請求超時控制
    CompletableFuture<LeaseGrantResponse> grant(long ttl, long timeout, TimeUnit unit);

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

  //主動續(xù)約1次
    CompletableFuture<LeaseKeepAliveResponse> keepAliveOnce(long leaseId);

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

     //自動續(xù)約
    CloseableClient keepAlive(long leaseId, StreamObserver<LeaseKeepAliveResponse> observer);
}

public interface Lock extends CloseableClient {

    //綁定租約申請指定名稱的鎖，搶鎖成功返回鎖秘鑰，
    CompletableFuture<LockResponse> lock(ByteSequence name, long leaseId);

    //持鎖秘鑰解鎖
    CompletableFuture<UnlockResponse> unlock(ByteSequence lockKey);
}

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.創(chuàng)建租約,并自動續(xù)約
            createLease();

            //3.執(zhí)行加鎖，并為鎖對應的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();
            }
            //如果是主動續(xù)約，則關閉續(xù)約的定時任務

            //刪除租約
            if (lockState.getLeaseId() != 0L) {
                leaseClient.revoke(lockState.getLeaseId());
            }
        } catch (InterruptedException | ExecutionException e) {
            //todo:異常處理
        }
        log.info("線程：{} 釋放鎖", Thread.currentThread().getName());
    }

    // 創(chuàng)建一個租約
    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);
            //自動續(xù)約
            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");
                }
            };
            // 設置自動續(xù)約
            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釋放鎖
    }
}