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);
网友评论