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 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 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 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() {

@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 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 observer = new StreamObserver() {

@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 async() {

return async(null);

}

@Override

public final RespT sync() {

return client.waitForCall(this::async);

}

GrpcClient

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

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