Java中使用etcd，包括基本的set、get、超时设置，watch监听等

etcd的使用文章。

etcd来zookeeper类似，常用的主要有set，get，getPrefix：获取指定前缀的所有数据，grant：key的超时设置，watch：监听回调事件，watchPrefix：监听某个前缀的事件，keepAlive：为某个key设置自动续约、自动刷新过期时间。

zk的大部分功能，etcd都有。但有一个，譬如虚拟节点，zk可以做到当客户端断开时，立马监听到，etcd要靠keepAlive续约机制，超过几秒不续约了，则认为掉线了。

还有etcd的超时时间，是需要先新建一个lease，再把这个lease在put时，作为参数传过去。

别的不多讲，直接上使用指南。

引入pom.xml，https://github.com/IBM/etcd-java  这是一个etcd的java客户端。

1<dependency>
2                <groupId>com.ibm.etcd</groupId>
3                <artifactId>etcd-java</artifactId>
4                <version>0.0.13</version>
5            </dependency>

自行下载etcd，并启动它。

定义一个常用的客户端工具类的接口

1import com.ibm.etcd.api.KeyValue;
2import com.ibm.etcd.client.kv.KvClient;
3
4import java.util.List;
5
6/**
7 * @author wuweifeng wrote on 2019-12-09
8 * @version 1.0
9 */
10public interface IConfigCenter {
11    /**
12     * 存入key，value
13     */
14    void put(String key, String value);
15
16    /**
17     * 存入key、value，和租约id
18     */
19    void put(String key, String value, long leaseId);
20
21    /**
22     * 存入key、value，和过期时间，单位是秒
23     */
24    void putAndGrant(String key, String value, long ttl);
25
26    /**
27     * 根据key，获取value
28     */
29    String get(String key);
30
31    /**
32     * 获取指定前缀的所有key-value
33     */
34    List<KeyValue> getPrefix(String key);
35
36    /**
37     * 监听key
38     */
39    KvClient.WatchIterator watch(String key);
40
41    /**
42     * 监听前缀为key的
43     */
44    KvClient.WatchIterator watchPrefix(String key);
45
46    /**
47     * 自动续约
48     * @param frequencySecs 续约频率，最小是4秒，默认是5秒
49     * @param minTtl 最小存活时间，最小是2秒，默认是10秒
50     * @return 返回leaseId
51     */
52    long keepAlive(String key, String value, int frequencySecs, int minTtl) throws Exception;
53
54    /**
55     * 判断剩余的过期时间
56     */
57    long timeToLive(long leaseId);
58}

实现这个接口

1import com.google.protobuf.ByteString;
2import com.ibm.etcd.api.KeyValue;
3import com.ibm.etcd.api.LeaseGrantResponse;
4import com.ibm.etcd.api.RangeResponse;
5import com.ibm.etcd.client.KvStoreClient;
6import com.ibm.etcd.client.kv.KvClient;
7import com.ibm.etcd.client.lease.LeaseClient;
8import com.ibm.etcd.client.lease.PersistentLease;
9import com.ibm.etcd.client.lock.LockClient;
10
11import org.springframework.util.CollectionUtils;
12
13import java.util.List;
14import java.util.concurrent.ExecutionException;
15
16import static java.util.concurrent.TimeUnit.SECONDS;
17
18/**
19 * etcd客户端
20 *
21 * @author wuweifeng wrote on 2019-12-06
22 * @version 1.0
23 */
24public class JdEtcdClient implements IConfigCenter {
25
26    private KvClient kvClient;
27    private LeaseClient leaseClient;
28    private LockClient lockClient;
29
30    public JdEtcdClient(KvStoreClient kvStoreClient) {
31        this.kvClient = kvStoreClient.getKvClient();
32        this.leaseClient = kvStoreClient.getLeaseClient();
33        this.lockClient = kvStoreClient.getLockClient();
34    }
35
36    @Override
37    public void put(String key, String value) {
38        kvClient.put(ByteString.copyFromUtf8(key), ByteString.copyFromUtf8(value)).sync();
39    }
40
41    @Override
42    public void put(String key, String value, long leaseId) {
43        kvClient.put(ByteString.copyFromUtf8(key), ByteString.copyFromUtf8(value), leaseId).sync();
44    }
45
46    @Override
47    public void putAndGrant(String key, String value, long ttl) {
48        LeaseGrantResponse lease = leaseClient.grant(ttl).sync();
49        put(key, value, lease.getID());
50    }
51
52    @Override
53    public String get(String key) {
54        RangeResponse rangeResponse = kvClient.get(ByteString.copyFromUtf8(key)).sync();
55        List<KeyValue> keyValues = rangeResponse.getKvsList();
56        if (CollectionUtils.isEmpty(keyValues)) {
57            return null;
58        }
59        return keyValues.get(0).getValue().toStringUtf8();
60    }
61
62    @Override
63    public List<KeyValue> getPrefix(String key) {
64        RangeResponse rangeResponse = kvClient.get(ByteString.copyFromUtf8(key)).asPrefix().sync();
65        return rangeResponse.getKvsList();
66    }
67
68    @Override
69    public KvClient.WatchIterator watch(String key) {
70        return kvClient.watch(ByteString.copyFromUtf8(key)).start();
71    }
72
73    @Override
74    public KvClient.WatchIterator watchPrefix(String key) {
75        return kvClient.watch(ByteString.copyFromUtf8(key)).asPrefix().start();
76    }
77
78    @Override
79    public long keepAlive(String key, String value, int frequencySecs, int minTtl) throws Exception {
80        //minTtl秒租期，每frequencySecs秒续约一下
81        PersistentLease lease = leaseClient.maintain().leaseId(SystemClock.now()).keepAliveFreq(frequencySecs).minTtl(minTtl).start();
82        long newId = lease.get(3L, SECONDS);
83        put(key, value, newId);
84        return newId;
85    }
86
87    @Override
88    public long timeToLive(long leaseId) {
89        try {
90            return leaseClient.ttl(leaseId).get().getTTL();
91        } catch (InterruptedException | ExecutionException e) {
92            e.printStackTrace();
93            return 0L;
94        }
95    }
96}

其中的SystemClock类，就在网上找一下吧，就是一个获取系统当前时间戳的工具类。

创建这个Client

1 /**
2     * @param endPoints 如https://127.0.0.1:2379 有多个时逗号分隔
3     */
4    public static JdEtcdClient build(String endPoints) {
5        return new JdEtcdClient(EtcdClient.forEndpoints(endPoints).withPlainText().build());
6    }

之后就可以使用这个etcd客户端工具类了。