etcd-java使用

简介

项目地址

https://github.com/IBM/etcd-java

特性

• Accepts/exposes protobuf-generated types directly to minimize overhead(请求头)
• Simpler fluent versions of KV primitives for making async or sync requests(fluent api的方式方便制定sync还是async)
• Automatic retries and (re)authentication
• Resilient(可快速恢复的) and scalable Lease and Watch abstractions with semantics aligned to most typical usage
  • Underlying stream RPCs only kept open for as long as they are needed
• Comes with some helpful utility classes with local cache, persistent key and leader election abstractions
• Simple declarative client configuration (optional) - encapsulate per-cluster connection details in a
  JSON document
• Support for connecting to multi-endpoint IBM Compose etcd deployments over TLS
• Currently doesn't cover all parts of the etcd-server API, in particular those related to cluster administration(集群管理) such as maintenance, cluster and auth management

kv操作

获取client

KvStoreClient client = EtcdClient.forEndpoint("localhost", 2379).withPlainText().build();

//KvStoreClient是一个接口，定义了如下三个client方法
KvClient kvClient = client.getKvClient();
LeaseClient leaseClient = client.getLeaseClient();
LockClient lockClient = client.getLockClient();

EtcdClient实现了Closeable接口的close方法，所以可以使用try-resource的语法，自动释放资源。示例代码见ClientBuilderTest

单个endpoint

try (KvStoreClient client = EtcdClient.forEndpoint("localhost", 2379).withPlainText().build()) {}

多个endpoint

try (KvStoreClient client = EtcdClient.forEndpoints(
                "localhost:2379,http://localhost:2379,https://localhost:2379,dns:///localhost:2379")
                .withPlainText().build()) {}

put操作

basic put

kvc.put(a, v2).sync();

batch put

kvc.batch().put(pr1).put(pr2).sync();

put with lease

//先创建lease
LeaseGrantResponse lgr = leaseClient.grant(2).sync();

//put的时候指定lease id
//如果lease不存在会抛出异常
kvc.put(bs("hello"), bs("world"), lgr.getID()).sync();

put后获取信息

获取上一次的值

• 如果key没有对应的值，返回的是KeyValue.getDefaultInstance()
• 如果之前key有对应的指，返回之前key对应的值

kvc.put(a, v2).prevKv().sync().getPrevKv();

获取header信息

kvc.put(a, v2).sync().getHeader();

get操作

basic get

同步get

RangeResponse rr = kvc.get(bs("a")).sync();

异步get

ListenableFuture<RangeResponse> rrFut1 = kvc.get(bs("new")).async();
rrFut1.get(1000, TimeUnit.SECONDS); //指定超时时间

batch get

TxnResponse txnResponse = kvc.batch().get(kvc.get(a).asRequest()).get(kvc.get(b).asRequest()).sync();
assertEquals(v1, txnResponse.getResponses(0).getResponseRange().getKvs(0).getValue());
assertEquals(v2, txnResponse.getResponses(1).getResponseRange().getKvs(0).getValue());

指定重试策略

ListenableFuture<RangeResponse> rrFut2 = kvc.get(bs("new"))
.deadline(Deadline.after(20, TimeUnit.SECONDS))
.backoffRetry().async(); // should work

delete操作

basic delete

• 获取删除的个数

kvc.delete(bs("notthere")).sync().getDeleted()

• 获取删除的key对应的value

kvc.delete(a).prevKv().sync().getPrevKvs(0).getValue()

batch delete

kvc.batch().delete(kvc.delete(a).asRequest())
            .delete(kvc.delete(b).asRequest()).sync();

transaction

ListenableFuture<TxnResponse> tresp = kvc.txnIf().cmpEqual(a).value(v1)
.and().cmpNotEqual(b).version(10)
.then().put(kvc.put(bs("new"), bs("newval")).asRequest()).async();

lease操作

普通lease

grant

LeaseGrantResponse lgr = leaseClient.grant(5L).sync(); //不指定leaseId
lgr.getID(); //获取id
lgr.getTTL(); //存活时间

leaseClient.grant(10L).leaseId(456L).sync(); //指定leaseId

keep-alive

leaseClient.keepAliveOnce(id).get(1, SECONDS)

revoke

leaseClient.revoke(id).get(1, SECONDS);
assertEquals(-1L, leaseClient.ttl(id).get().getTTL());//撤销后ttl变成-1

persistent lease

basic

PersistentLease pl2 = lc.maintain().minTtl(minTtl).keepAliveFreq(kaFreq).start();
long newLeaseId = pl2.get(20, SECONDS);

with StreamObserver

PersistentLease pl = lc.maintain().minTtl(minTtl).keepAliveFreq(kaFreq)
                    .start(new StreamObserver<LeaseState>() {
                        @Override
                        public void onNext(LeaseState value) {
                            System.out.println(t(start) + "PL state change: " + value);
                            observerEvents.add(value);
                        }
                        @Override
                        public void onError(Throwable t) {
                            System.out.println(t(start) + "PL error: " + t);
                            observerEvents.add(t);
                        }
                        @Override
                        public void onCompleted() {
                            System.out.println(t(start) + "PL completed");
                            observerEvents.add(COMPLETED);
                        }
                    });

list client lease

需要注意的是persistent lease是基于client的会话的，如果client挂了的话，lease是会自动释放的

Set<Long> lidsFound = client2.getLeaseClient().list().get()
                    .getLeasesList().stream().map(LeaseStatus::getID).collect(Collectors.toSet());

lock操作

同一个客户端lock不同的lease会报错，详细见LockTest.testWithContention

普通lock

LockResponse lr = lockClient.lock(KeyUtils.bs("mylock"))
        .deadline(Deadline.after(500, TimeUnit.MILLISECONDS)).sync(); //获取锁
assertTrue(kvClient.txnIf().exists(lr.getKey()).sync().getSucceeded());

assertNotNull(lockClient.unlock(lockKey).sync()); //释放锁
assertFalse(kvClient.txnIf().exists(lockKey).sync().getSucceeded());

lock使用lease

LockResponse lr = lockClient.lock(KeyUtils.bs("mylock2")).withLease(lgr.getID()).sync();

lock使用persistent lease

PersistentLease pl = lec.maintain().start();
LockResponse lr = lockClient.lock(KeyUtils.bs("mylock3")).withLease(pl).sync();
//如果pl.close后会自动释放

watch操作

简单watch

WatchIterator watchIterator = kvc.watch(bs("/watchtest")).asPrefix().start();

watch指定observer

//创建一个observer
final StreamObserver<WatchUpdate> observer = new StreamObserver<WatchUpdate>() {
                @Override
                public void onNext(WatchUpdate value) {
                    System.out.println(t(start) + "watch event: " + value);
                    watchEvents.add(value);
                }
                @Override
                public void onError(Throwable t) {
                    System.out.println(t(start) + "watch error: " + t);
                    watchEvents.add(t);
                }
                @Override
                public void onCompleted() {
                    System.out.println(t(start) + "watch completed");
                    watchEvents.add(COMPLETED);
                }
            };

//启动watcher，并且绑定observer
Watch watch = kvc.watch(bs("/watchtest")).asPrefix().start(observer);

获取watch的事件

WatchUpdate wu = getNextSkipEmpty(watchEvents);
assertNotNull(wu);
assertEquals("event: " + wu, 1, wu.getEvents().size()); //事件个数
assertEquals(EventType.PUT, wu.getEvents().get(0).getType()); //事件类型
assertEquals(bs("a value"), wu.getEvents().get(0).getKv().getValue()); //值
assertEquals(bs("/watchtest/a"), wu.getEvents().get(0).getKv().getKey()); //键

election

简单的case参见示例代码LeaderElectionTest

源码浅析

EtcdClient

KvStoreClient.png

重点方法是line397行，EtcdClient的构造方法，初始化了internalExecutor，channel，grpc，kvClient等

EtcdKvClient

etcd kv操作和watch操作的核心类，实现了KvClient接口，其他的leaseClient和lockClient类似

EtcdKvClient.png

重点关注AbstractFluentRequest，底层的client通过grpc进行远程rpc调用。
所有的操作在底层抽象为sync和async的操作,sync操作其实也是async操作，只是在client层做了get操作。

@Override
public final ListenableFuture<RespT> async() {
    return async(null);
}

@Override
public final RespT sync() {
    return client.waitForCall(this::async);
}

GrpcClient

grpc调用的封装类, 底层调用方法

ClientCalls.futureUnaryCall(channel.newCall(method, callOptions), request)