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)
