etcd入门之jetcd

java项目可以通过jetcd客户端使用etcd提供的分布式协调服务，让java项目分布式协调组件除了zookeeper外有了另一个选择。zookeeper与etcd的优劣在网上可以找到不少的分析文章。

1 maven依赖

java工程使用jetcd只需要添加相应的依赖（0.6.x版本要求JDK的最低版本是JDK 11，0.5.x和0.7.x版本支持JDK 8，大家实际使用中可以根据自己使用的JDK版本选择）

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

如果使用上面的版本运行起来后报netty 某个方法不存在的错误，可能是netty版本的问题，尝试引入相应版本的netty依赖，经测试，以下netty和jetcd版本组合在JDK 8中可以正常使用。

    <!-- etcd -->
    <dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-buffer</artifactId>
        <version>4.1.66.Final</version>
        <exclusions>
            <exclusion>
                <groupId>io.netty</groupId>
                <artifactId>netty-common</artifactId>
            </exclusion>
        </exclusions>
    </dependency>

    <dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-common</artifactId>
        <version>4.1.66.Final</version>
    </dependency>

    <dependency>
        <groupId>io.etcd</groupId>
        <artifactId>jetcd-core</artifactId>
        <version>0.7.3</version>
        <exclusions>
            <exclusion>
                <groupId>io.netty</groupId>
                <artifactId>netty-buffer</artifactId>
            </exclusion>
            <exclusion>
                <groupId>io.netty</groupId>
                <artifactId>netty-common</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
        

2 连接etcd集群

通过以下语句可以获取代理etcd集群连接和操作的Client实例，endpoints就是etcd集群的listen-client-urls列表

    String[] endpoints = new String[]{"http://127.0.0.1:2379", "http://127.0.0.2:2379",  "http://127.0.0.2:2379"};
    Client etcdClient = Client.builder().endpoints(endpoints).build();

如果需要用户名和密码验证，可以通过user()和password()参数指定

    String[] endpoints = new String[]{"http://127.0.0.1:2379", "http://127.0.0.2:2379",  "http://127.0.0.2:2379"};
    String user = "test";
    String passwd = "testpw";
    Client etcdClient = Client.builder().endpoints(endpoints).user(user).password(passwd).build();

同样的，如果通过TLS协议与etcd通信，可以通过sslContext()指定TLS证书信息

    String[] endpoints = new String[]{"http://127.0.0.1:2379", "http://127.0.0.2:2379",  "http://127.0.0.2:2379"};
    SslContext sslContext = ;
    Client etcdClient = Client.builder().endpoints(endpoints).sslContext(sslContext).build();

有了Client对象，就可以获取etcd的具体操作对象，与etcd集群进行交互了。如：

    // 集群管理
    Cluster etcdCluster = etcdClient.getClusterClient();
    // 操作键值对
    KV kv = etcdClient.getKVClient();
    // 操作租约
    Lease lease = etcdClient.getLeaseClient();
    // 监听功能
    Watch watch = etcdClient.getWatchClient();
    // 选主
    Election election = etcdClient.getElectionClient();
    // 分布式锁
    Lock lock = etcdClient.getLockClient();

3 集群管理

    Cluster etcdCluster = etcdClient.getClusterClient();
    
    // 获取集群所有成员
    CompletableFuture<MemberListResponse> memberResponse = etcdCluster.listMember();
    // 阻塞获取成员信息
    List<Member> members = memberResponse.get().getMembers();
    for (Member member : members) {
        // 包括集群每个节点的id、name，发布的peer url和client url
        log.info("memberId: {}, memberName: {}, peerUrls: {}, clientUrls: {}", member.getId(), member.getName(), member.getPeerURIs(), member.getClientURIs());
    }

通过Cluster还可以向etcd集群进行添加和删除节点等操作

4 KV操作

    KV kv = etcdClient.getKVClient();
    
    ByteSequence key = ByteSequence.from("test-key",StandardCharsets.UTF_8);
    ByteSequence value = ByteSequence.from("test-value",StandardCharsets.UTF_8);
    
    // 写入key value
    CompletableFuture<PutResponse> putFuture = kv.put(key, value);
    
    // 读key的值
    CompletableFuture<GetResponse> getFuture = kv.get(key);
    
    // 阻塞获取，值应该为上面写入的test-value
    String retValue = getFuture.get().getKvs().get(0).getValue().toString(StandardCharsets.UTF_8);
    
    // 前缀查询
    GetOption prefixOption = GetOption.newBuilder().isPrefix(true).build();
    CompletableFuture<GetResponse> getFuture = kv.get(key, prefixOption);
    
    // 范围查询
    ByteSequence endKey = ByteSequence.from("test-key100",StandardCharsets.UTF_8);
    GetOption rangeOption = GetOption.newBuilder().withRange(endKey).build();
    CompletableFuture<GetResponse> getFuture = kv.get(key, rangeOption);    

5 lease操作

    Lease lease = etcdClient.getLeaseClient();
    long leaseId = 0L;
    // 创建一个租约，租约TTL时间为60秒
    leaseId = lease.grant(60).get().getID();
    
    // 写入key-value值时绑定一个租约，租约到期后，key会被删除
    KV kv = etcdClient.getKVClient();
    PutOption option = PutOption.newBuilder().withLeaseId(leaseId).build();
    CompletableFuture<PutResponse> putFuture = kv.put(key, value, option);
    
    // 获取租约信息，比如租约id，授予的ttl，剩余的ttl，哪些key绑定了改租约
    CompletableFuture<LeaseTimeToLiveResponse> leaseFuture= lease.timeToLive(leaseId, leaseOption);
    
    // 为租约续约一次
    lease.keepAliveOnce(leaseId);
    
    // 持续续约，在租约的三分之一时间时，就会自动向etcd续约。相当于租约到期前会执行三次keepAliveOnce()
    lease.keepAlive(leaseId, observer);
    
    // 主动撤销租约
    lease.revoke(leaseId);

执行lease.keepAlive()方法持续向etcd集群续约时，可以提供一个observer，跟踪续约情况：

    public static class ClusterNodeKeepAliveObserver implements 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: {}", ExceptionUtils.getStackTrace(t));
        }

        @Override
        public void onCompleted() {
            log.trace("cluster node lease completed");
        }
    }

6 watch

可以监听某个key，并注册key发生变化的处理方法

    Watch watch = etcdClient.getWatchClient();
    ByteSequence key = ByteSequence.from("test-key",StandardCharsets.UTF_8);
    watch.watch(key, watchResponse -> {
    
        watchResponse.getEvents().forEach(event -> {

            switch (event.getEventType()) {

                case DELETE:
                    // 当KEY被删除触发
                    break;
                case PUT:
                    // 当key被创建或修改
                    break;
                default:
                    break;
            }
        });;
    });

当然也支持前缀监控、范围监控等监控形式

    Watch watch = etcdClient.getWatchClient();
    ByteSequence key = ByteSequence.from("test-key",StandardCharsets.UTF_8);
    // 前缀
    WatchOption option = WatchOption.newBuilder().isPrefix(true).build();
    // 范围
    // ByteSequence endKey = ByteSequence.from("test-key100",StandardCharsets.UTF_8);
    // WatchOption option = WatchOption.newBuilder().withRange(endKey).build();
    watch.watch(key, option, watchResponse -> {
    
        watchResponse.getEvents().forEach(event -> {

            switch (event.getEventType()) {

                case DELETE:
                    // 当KEY被删除触发
                    break;
                case PUT:
                    // 当key被创建或修改
                    break;
                default:
                    break;
            }
        });;
    });

7 election选主

    Election election = etcdClient.getElectionClient();
    // 代表某个选主活动的key，抢到主的节点会在election key目录下创建一个主节点key
    ByteSequence electionKey = ByteSequence.from("/election", StandardCharsets.UTF_8);
    
    // 节点抢到主时，写入上述创建节点key的值，一般是能标识主节点的唯一标识符
    ByteSequence proposal = ByteSequence.from("member01", StandardCharsets.UTF_8);
    
    // 主节点的租约，租约到期的话，主节点key被删除，主节点失去主地位，其他节点可以重新抢主
    Lease lease = etcdClient.getLeaseClient();
    long leaseId = lease.grant(60).get().getID();
    // 抢到了主，持续保活
    lease.keepAlive(leaseId, new ClusterNodeKeepAliveObserver());
    
    // 竞选主    
    CompletableFuture<CampaignResponse> campaign = election.campaign(electionKey, leaseId, proposal);
    
    // 抢到了主，则正常返回，Leader key可以获取到主节点的key值等信息
    // 没抢到主，则一直阻塞（因为这里用的是阻塞获取，也可以异步获取），或抛出异常
    LeaderKey key = campaign.get().getLeader();

也可以为election设置一个observer，这样主发生变化时，可以收到通知，从而进行相应处理，比如跟踪主的变化，谁是当前的主等

    election.observe(electionKey, new LeaderElectionListener());

    public class LeaderElectionListener implements Election.Listener {

        @Override
        public void onNext(LeaderResponse response) {
            String newLeader = response.getKv().getValue().toString(StandardCharsets.UTF_8);
            log.info("leader is changed. new: {}", newLeader);
        }

        @Override
        public void onError(Throwable t) {
            log.error("leader election failed. exception info: {}", ExceptionUtils.getStackTrace(t));
        }

        @Override
        public void onCompleted() {
            log.trace("leader election completed");
        }
    }

8 分布式lock

    Lock lock = etcdClient.getLockClient();
    
    // 获取到锁的节点租约，租约到期的话，key被删除，其他节点可以重新获取锁
    Lease lease = etcdClient.getLeaseClient();
    long leaseId = lease.grant(60).get().getID();
    // 持续保活
    lease.keepAlive(leaseId, new ClusterNodeKeepAliveObserver());
    
    ByteSequence lockKey = ByteSequence.from("lock_key",StandardCharsets.UTF_8);
    // 尝试获取锁
    CompletableFuture<LockResponse> lockFuture = lock.lock(lockKey, leaseId);
    // 获取到分布式锁，则返回（也可以异步获取锁通知）
    String lockStr = lockFuture.get().getKey().toString(StandardCharsets.UTF_8);