<dependency>

  <groupId>io.etcd</groupId>

  <artifactId>jetcd-core</artifactId>

  <version>${jetcd-version}</version>

</dependency>

// 创建客户端，本例中Etcd服务端为单机模式

Client client = Client.builder().endpoints("http://localhost:2379").build();

KV kvClient = client.getKVClient();

// 对String类型的Key-Value进行类型转换

ByteSequence key = ByteSequence.fromString("test_key");

ByteSequence value = ByteSequence.fromString("test_value");



// put操作，等待操作完成

kvClient.put(key, value).get();



// get操作，等待操作完成

CompletableFuture<GetResponse> getFuture = kvClient.get(key);

GetResponse response = getFuture.get();



// delete操作，等待操作完成

kvClient.delete(key).get();

// 创建客户端，本例中Etcd服务端为单机模式

Client client = Client.builder().endpoints("http://localhost:2379").build();



 // 创建Lease客户端

Lease leaseClient = client.getLeaseClient();



// 创建一个60s的租约，等待完成，超时设置阈值30s

Long leaseId = leaseClient.grant(60).get(30, TimeUnit.SECONDS).getID();



// 使指定的租约永久有效，即永久租约

leaseClient.keepAlive(leaseId);



// 续约一次

leaseClient.keepAliveOnce(leaseId);



// 解除租约，绑定该租约的键值将被删除

leaseClient.revoke(leaseId);



// 检索指定ID对应的租约的详细信息

LeaseTimeToLiveResponse lTRes = leaseClient.timeToLive(leaseId, LeaseOption.newBuilder().withAttachedKeys().build()).get(); 

// 创建客户端，本例中Etcd服务端为单机模式

Client client = Client.builder().endpoints("http://localhost:2379").build();



// 对String类型的Key进行类型转换

ByteSequence key = ByteSequence.fromString("test_key");



// 创建Watch客户端

Watch watchClient = client.getWatchClient();

// 为Key创建一个Watcher

Watcher watcher = watch.watch(key);



// 开始listen，如果监听的Key有事件(如删除、更新等)发生则返回

WatchResponse response = null;

try

{

    response = watcher.listen();

}

catch (InterruptedException e)

{

    System.out.println("Failed to listen key:"+e);

}

if(response != null)

{

    List<WatchEvent> eventlist = res.getEvents();

    // 解析eventlist，判断是否为自己关注的事件，作进一步操作

    // To do something

}   

// 创建客户端，本例中Etcd服务端为单机模式

Client client = Client.builder().endpoints("http://localhost:2379").build();



// 创建Cluster客户端

Cluster clusterClient = client.getClusterClient();



// 获取集群成员列表

List<Member> list = clusterClient.listMember().get().getMembers();



// 向集群中添加成员

String tempAddr = "http://localhost:2389";

List<String> peerAddrs = new ArrayList<String>();

peerAddrs.add(tempAddr); 

clusterClient.addMember(peerAddrs);



// 根据成员ID删除成员

long memberID = 8170086329776576179L;

clusterClient.removeMember(memberID);



// 更新

clusterClient.updateMember(memberID, peerAddrs);

// 创建客户端，本例中Etcd服务端为单机模式

Client client = Client.builder().endpoints("http://localhost:2379").build();



// 创建一个Maintenance 客户端

Maintenance maintClient = client.getMaintenanceClient();



// 获取指定节点的状态,对res做进一步解析可得节点状态详情

StatusResponse res = maintClient.statusMember("http://localhost:2379").get();



// 对指定的节点进行碎片整理，在压缩键空间之后，后端数据库可能呈现内部碎片，需进行整理

// 整理碎片是一个“昂贵”的操作，应避免同时对多个节点进行整理

maintClient.defragmentMember("http://localhost:2379").get();



// 获取所有活跃状态的键空间告警

List<AlarmMember> alarmList = maintClient.listAlarms().get().getAlarms();



// 解除指定键空间的告警

maintClient.alarmDisarm(alarmList.get(0));

import java.util.List;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.Executors;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.TimeoutException;



import com.coreos.jetcd.Client;

import com.coreos.jetcd.KV;

import com.coreos.jetcd.Lease;

import com.coreos.jetcd.Watch.Watcher;

import com.coreos.jetcd.options.GetOption;

import com.coreos.jetcd.options.GetOption.SortTarget;

import com.coreos.jetcd.options.PutOption;

import com.coreos.jetcd.watch.WatchEvent;

import com.coreos.jetcd.watch.WatchResponse;

import com.coreos.jetcd.data.ByteSequence;

import com.coreos.jetcd.data.KeyValue;

import com.coreos.jetcd.kv.PutResponse;



import java.util.UUID;



/**

 * Etcd 客户端代码，用多个线程“抢锁”模拟分布式系统中，多个进程“抢锁”

 *

 */

public class EtcdClient

{



    public static void main(String[] args) throws InterruptedException, ExecutionException,

            TimeoutException, ClassNotFoundException

    {

            // 创建Etcd客户端，Etcd服务端为单机模式

        Client client = Client.builder().endpoints("http://localhost:2379").build();



        // 对于某共享资源制定的锁名

        String lockName = "/lock/mylock";



        // 模拟分布式场景下，多个进程“抢锁”

        for (int i = 0; i < 3; i++)

        {

            new MyThread(lockName, client).start();

        }    

    }



    /**

     * 加锁方法，返回值为加锁操作中实际存储于Etcd中的Key，即：lockName+UUID，

     * 根据返回的Key，可删除存储于Etcd中的键值对，达到释放锁的目的。

     * 

     * @param lockName

     * @param client

     * @param leaseId

     * @return

     */

    public static String lock(String lockName, Client client, long leaseId)

    {

        // lockName作为实际存储在Etcd的中的Key的前缀，后缀是一个全局唯一的ID，从而确保：对于同一个锁，不同进程存储的Key具有相同的前缀，不同的后缀

        StringBuffer strBufOfRealKey = new StringBuffer();

        strBufOfRealKey.append(lockName);

        strBufOfRealKey.append("/");

        strBufOfRealKey.append(UUID.randomUUID().toString());



        // 加锁操作实际上是一个put操作，每一次put操作都会使revision增加1，因此，对于任何一次操作，这都是唯一的。(get,delete也一样)

        // 可以通过revision的大小确定进行抢锁操作的时序，先进行抢锁的，revision较小，后面依次增加。

        // 用于记录自己“抢锁”的Revision，初始值为0L

        long revisionOfMyself = 0L;



        KV kvClient = client.getKVClient();

        // lock，尝试加锁，加锁只关注Key，value不为空即可。

        // 注意：这里没有考虑可靠性和重试机制，实际应用中应考虑put操作而重试

        try

        {

            PutResponse putResponse = kvClient

                    .put(ByteSequence.fromString(strBufOfRealKey.toString()),

                            ByteSequence.fromString("value"),

                            PutOption.newBuilder().withLeaseId(leaseId).build())

                    .get(10, TimeUnit.SECONDS);



            // 获取自己加锁操作的Revision号

            revisionOfMyself = putResponse.getHeader().getRevision();

        }

        catch (InterruptedException | ExecutionException | TimeoutException e1)

        {

            System.out.println("[error]: lock operation failed:" + e1);

        }



        try

        {

            // lockName作为前缀，取出所有键值对，并且根据Revision进行升序排列，版本号小的在前

            List<KeyValue> kvList = kvClient.get(ByteSequence.fromString(lockName),

                    GetOption.newBuilder().withPrefix(ByteSequence.fromString(lockName))

                            .withSortField(SortTarget.MOD).build())

                    .get().getKvs();



            // 如果自己的版本号最小，则表明自己持有锁成功，否则进入监听流程，等待锁释放

            if (revisionOfMyself == kvList.get(0).getModRevision())

            {

                System.out.println("[lock]: lock successfully. [revision]:" + revisionOfMyself);

                // 加锁成功，返回实际存储于Etcd中的Key

                return strBufOfRealKey.toString();

            }

            else

            {

                // 记录自己加锁操作的前一个加锁操作的索引，因为只有前一个加锁操作完成并释放，自己才能获得锁

                int preIndex = 0;

                for (int index = 0; index < kvList.size(); index++)

                {                 

                    if (kvList.get(index).getModRevision() == revisionOfMyself)

                    {

                        preIndex = index - 1;// 前一个加锁操作，故比自己的索引小1

                    }

                }

                // 根据索引，获得前一个加锁操作对应的Key

                ByteSequence preKeyBS = kvList.get(preIndex).getKey();



                // 创建一个Watcher，用于监听前一个Key

                Watcher watcher = client.getWatchClient().watch(preKeyBS);

                WatchResponse res = null;

                // 监听前一个Key，将处于阻塞状态，直到前一个Key发生delete事件

                // 需要注意的是，一个Key对应的事件不只有delete，不过，对于分布式锁来说，除了加锁就是释放锁

                // 因此，这里只要监听到事件，必然是delete事件或者Key因租约过期而失效删除，结果都是锁被释放

                try

                {

                        System.out.println("[lock]: keep waiting until the lock is released.");

                    res = watcher.listen();

                }

                catch (InterruptedException e)

                {

                    System.out.println("[error]: failed to listen key.");

                }



                // 为了便于读者理解，此处写一点冗余代码，判断监听事件是否为DELETE，即释放锁

                List<WatchEvent> eventlist = res.getEvents();

                for (WatchEvent event : eventlist)

                {

                        // 如果监听到DELETE事件，说明前一个加锁操作完成并已经释放，自己获得锁，返回

                    if (event.getEventType().toString().equals("DELETE"))

                    {

                            System.out.println("[lock]: lock successfully. [revision]:" + revisionOfMyself);

                        return strBufOfRealKey.toString();

                    }

                }

            }

        }

        catch (InterruptedException | ExecutionException e)

        {

            System.out.println("[error]: lock operation failed:" + e);

        }



        return strBufOfRealKey.toString();

    }



    /**

     * 释放锁方法，本质上就是删除实际存储于Etcd中的Key

     * 

     * @param lockName

     * @param client

     */

    public static void unLock(String realLockName, Client client)

    {

        try

        {

            client.getKVClient().delete(ByteSequence.fromString(realLockName)).get(10,

                    TimeUnit.SECONDS);

            System.out.println("[unLock]: unlock successfully.[lockName]:" + realLockName);

        }

        catch (InterruptedException | ExecutionException | TimeoutException e)

        {

            System.out.println("[error]: unlock failed：" + e);

        }

    }



    /**

     * 自定义一个线程类，模拟分布式场景下多个进程 "抢锁"

     */

    public static class MyThread extends Thread

    {

        private String lockName;

        private Client client;



        MyThread(String lockName, Client client)

        {

            this.client = client;

            this.lockName = lockName;

        }



        @Override

        public void run()

        {

            // 创建一个租约，有效期15s

            Lease leaseClient = client.getLeaseClient();

            Long leaseId = null;

            try

            {

                leaseId = leaseClient.grant(15).get(10, TimeUnit.SECONDS).getID();

            } 

            catch (InterruptedException | ExecutionException | TimeoutException e1)

            {

                 System.out.println("[error]: create lease failed:" + e1);

                 return;

            }



            // 创建一个定时任务作为“心跳”，保证等待锁释放期间，租约不失效；

            // 同时，一旦客户端发生故障，心跳便会中断，锁也会应租约过期而被动释放，避免死锁

                ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();  

            // 续约心跳为12s，仅作为举例  

            service.scheduleAtFixedRate(new KeepAliveTask(leaseClient, leaseId), 1, 12, TimeUnit.SECONDS); 



            // 1. try to lock

            String realLoclName = lock(lockName, client, leaseId);



            // 2. to do something

            try

            {

                Thread.sleep(6000);

            }

            catch (InterruptedException e2)

            {

                System.out.println("[error]:" + e2);

            }

            // 3. unlock

            service.shutdown();// 关闭续约的定时任务

            unLock(realLoclName, client);

        }

    }



    /**

     * 在等待其它客户端释放锁期间，通过心跳续约，保证自己的Key-Value不会失效

     *

     */

    public static class KeepAliveTask implements Runnable

    {

            private Lease leaseClient;

            private long leaseId;



            KeepAliveTask(Lease leaseClient, long leaseId)

            {

                this.leaseClient = leaseClient;

                this.leaseId = leaseId;

            }



            @Override

        public void run()

        {

                leaseClient.keepAliveOnce(leaseId);     

        }

    }

}

[lock]: lock successfully. [revision]:44

[lock]: keep waiting until the lock is released.

[lock]: keep waiting until the lock is released.

[unLock]: unlock successfully.

[lock]: lock successfully. [revision]:45

[unLock]: unlock successfully.

[lock]: lock successfully. [revision]:46

[unLock]: unlock successfully.

export ETCDCTL_API=3

./etcdctl.exe lock mylock

$ ./etcdctl.exe lock mylock

mylock/694d65eb367c7ec4

$ ./etcdctl.exe lock mylock

mylock/694d65eb367c7ec8

// 创建客户端，本例中Etcd服务端为单机模式

Client client = Client.builder().endpoints("http://localhost:2379").build();



// 创建Lock客户端

Lock lockClient = client.getLockClient();



// 创建Lease客户端，并创建一个有效期为30s的租约

Lease leaseClient = client.getLeaseClient()；

long leaseId = leaseClient.grant(30).get().getID();



// 加、解锁操作

try

{

    // 调用lock接口，加锁，并绑定租约

    lockClient.lock(ByteSequence.fromString("lockName"), leaseId).get();

    // 调用unlock接口，解锁

    lockClient.unlock(ByteSequence.fromString(lockName)).get();

}

catch (InterruptedException | ExecutionException e1)

{

    System.out.println("[error]: lock failed:" + e1);

}

import java.util.concurrent.ExecutionException;

import java.util.concurrent.Executors;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;



import com.coreos.jetcd.Client;

import com.coreos.jetcd.Lease;

import com.coreos.jetcd.Lock;

import com.coreos.jetcd.data.ByteSequence;



public class DistributedLock

{

    private static DistributedLock lockProvider = null;

    private static Object mutex = new Object();

    private Client client;

    private Lock lockClient;

    private Lease leaseClient;



    private DistributedLock()

    {

        super();

        // 创建Etcd客户端，本例中Etcd集群只有一个节点

        this.client = Client.builder().endpoints("http://localhost:2379").build();

        this.lockClient = client.getLockClient();

        this.leaseClient = client.getLeaseClient();

    }



    public static DistributedLock getInstance()

    {

        synchronized (mutex)

        {

            if (null == lockProvider)

            {

                lockProvider = new DistributedLock();

            }

        }

        return lockProvider;

    }



    /**

     * 加锁操作，需要注意的是，本例中没有加入重试机制，加锁失败将直接返回。

     * 

     * @param lockName: 针对某一共享资源(数据、文件等)制定的锁名

     * @param TTL : Time To Live，租约有效期，一旦客户端崩溃，可在租约到期后自动释放锁

     * @return LockResult

     */

    public LockResult lock(String lockName, long TTL)

    {

        LockResult lockResult = new LockResult();

        /*1.准备阶段*/

        // 创建一个定时任务作为“心跳”，保证等待锁释放期间，租约不失效；

        // 同时，一旦客户端发生故障，心跳便会停止，锁也会因租约过期而被动释放，避免死锁

        ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();



        // 初始化返回值lockResult

        lockResult.setIsLockSuccess(false);

        lockResult.setService(service);



        // 记录租约ID，初始值设为 0L

        Long leaseId = 0L;



        /*2.创建租约*/

        // 创建一个租约，租约有效期为TTL，实际应用中根据具体业务确定。

        try

        {

            leaseId = leaseClient.grant(TTL).get().getID();

            lockResult.setLeaseId(leaseId);



            // 启动定时任务续约，心跳周期和初次启动延时计算公式如下，可根据实际业务制定。

            long period = TTL - TTL / 5;

            service.scheduleAtFixedRate(new KeepAliveTask(leaseClient, leaseId), period, period,

                    TimeUnit.SECONDS);

        }

        catch (InterruptedException | ExecutionException e)

        {

            System.out.println("[error]: Create lease failed:" + e);

            return lockResult;

        }



        System.out.println("[lock]: start to lock." + Thread.currentThread().getName());



        /*3.加锁操作*/

        // 执行加锁操作，并为锁对应的Key绑定租约

        try

        {

            lockClient.lock(ByteSequence.fromString(lockName), leaseId).get();

        }

        catch (InterruptedException | ExecutionException e1)

        {

            System.out.println("[error]: lock failed:" + e1);

            return lockResult;

        }

        System.out.println("[lock]: lock successfully." + Thread.currentThread().getName());



        lockResult.setIsLockSuccess(true);



        return lockResult;

    }



    /**

     * 解锁操作，释放锁、关闭定时任务、解除租约

     * 

     * @param lockName:锁名

     * @param lockResult:加锁操作返回的结果

     */

    public void unLock(String lockName, LockResult lockResult)

    {

        System.out.println("[unlock]: start to unlock." + Thread.currentThread().getName());

        try

        {

            // 释放锁   

            lockClient.unlock(ByteSequence.fromString(lockName)).get();

            // 关闭定时任务

            lockResult.getService().shutdown();

            // 删除租约

            if (lockResult.getLeaseId() != 0L)

            {

                leaseClient.revoke(lockResult.getLeaseId());

            }

        }

        catch (InterruptedException | ExecutionException e)

        {

            System.out.println("[error]: unlock failed: " + e);

        }



        System.out.println("[unlock]: unlock successfully." + Thread.currentThread().getName());

    }



    /**

     * 在等待其它客户端释放锁期间，通过心跳续约，保证自己的锁对应租约不会失效

     *

     */

    public static class KeepAliveTask implements Runnable

    {

        private Lease leaseClient;

        private long leaseId;



        KeepAliveTask(Lease leaseClient, long leaseId)

        {

            this.leaseClient = leaseClient;

            this.leaseId = leaseId;

        }



        @Override

        public void run()

        {

            // 续约一次

            leaseClient.keepAliveOnce(leaseId);

        }

    }



    /**

     * 该class用于描述加锁的结果，同时携带解锁操作所需参数

     * 

     */

    public static class LockResult

    {

        private boolean isLockSuccess;

        private long leaseId;

        private ScheduledExecutorService service;



        LockResult()

        {

            super();

        }



        public void setIsLockSuccess(boolean isLockSuccess)

        {

            this.isLockSuccess = isLockSuccess;

        }



        public void setLeaseId(long leaseId)

        {

            this.leaseId = leaseId;

        }



        public void setService(ScheduledExecutorService service)

        {

            this.service = service;

        }



        public boolean getIsLockSuccess()

        {

            return this.isLockSuccess;

        }



        public long getLeaseId()

        {

            return this.leaseId;

        }



        public ScheduledExecutorService getService()

        {

            return this.service;

        }

    }



}