ETCD数据监听

系统管理员 2022-01-14 19:31 276阅读 0赞

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][]

ETCD

更新/删除/重新设置键的ttl 都会触发watcher , 但是如果在body中增加refresh=true ， 更新ttl(必须存在) ， 将不会出发watcher事件。

？wait=true 　　　　监听当前节点  
？recursive=true 　　 递归监听当前节点和子目录  
？waitIndex=xxx　　 监听过去已经发生的。过去值的查询或监听， 必选与wait一起使用。

watch 一个ttl自删除的key时，收到如下 “expire” action。

{
         "action": "expire",
         "node": {
             "createdIndex": 2223,
             "key": "/message",
             "modifiedIndex": 2224
         },
         "prevNode": {
             "createdIndex": 2223,
             "expiration": "2018-11-12T09:25:00.028597482Z",
             "key": "/message",
             "modifiedIndex": 2223,
             "value": ""
         }
     }

GET 对过去的键值操作进行查询：类似上面提到的监控，在其基础上指定过去某次修改的索引编号，就可以查询历史操作。默认可查询的历史记录为1000条。  
？waitIndex=xxx 监听过去已经发生的。 这个在确保在watch命令中，没有丢失事件非常有用。例如：我们反复watch 我们得到节点的 modifiedIndex+1。

因为 node的modifiedIndex的值是不连续，如果waitIndex的值没有相应modifiedIndex，返回比它大的最近的modifedIndex的节点信息。 如果大于节点中所有的modifiedIndex，等待，直到节点的modifiedIndex值大于等于waitIndex的值。

即使删除key后，也可以查询历史数据。

store中有一个全局的currentIndex，每次变更，index会加1.然后每个event都会关联到currentIndex.

当客户端调用watch接口（参数中增加 wait参数）时，如果监听的waitIndex 不存在与key对应的EventHistroy 中（currentIndex >= waitIndex），并且key对应的modifyIndex > waitIndex , 则会查找到第一个大于waitIndex 的数据进行展示。  
如果历史表中没有或者请求没有带 waitIndex，则放入WatchHub中，每个key会关联一个watcher列表。 当有变更操作时，变更生成的event会放入EventHistroy表中，同时通知和该key相关的watcher。

注意：  
1. 必须与 wait 一起使用；  
2. curl 中url需要使用引号。  
3. etcd 仅仅保留系统中所有key最近的1000条event，建议将获取到的response发送到另一个线程处理，而不是处理response而阻塞watch。  
4. 如果watch超出了etcd保存的最近1000条，建议get后使用response header中的 X-Etcd-Index + 1进行重新watch，而不是使用node中的modifiedIndex+1. 因为 X-Etcd-Index 永远大于等于modifiedIndex， 使用modifiedIndex可能会返回401错误码，同样超出。  
5. long polling可能会被服务器关闭，如超时或服务器关闭。导致仅仅收到header 200OK，body为空，此时应重新watch。

etcd4j 测试案例：

import mousio.etcd4j.EtcdClient;
    import mousio.etcd4j.promises.EtcdResponsePromise;
    import mousio.etcd4j.responses.EtcdKeysResponse;
    import org.junit.Before;
    import org.junit.Test;
    
    import java.io.IOException;
    import java.net.URI;
    import java.util.List;
    
    /**
     * @Description
     * @auther bozhu
     * @create 11\ 12\ 2018
     */
    public class EtcdClientAddListenerTest {
        EtcdClient client = null;
    
        @Before
        public void executeBefore() {
            client = new EtcdClient(URI.create("http://172.16.85.20:2379"));
        }
    
        private String getConfig(String configFile , EtcdKeysResponse dataTree) {
            List<EtcdKeysResponse.EtcdNode> nodes = null;
            if(null != dataTree ) {
                return dataTree.getNode().getValue();
            }
            System.out.println("Etcd configFile"+ configFile+"is not exist,Please Check");
            return null;
        }
        @Test
        public void testStask() throws Exception {
            for (int i = 1; i < 5000; i++) {
                EtcdKeysResponse etcdKeysResponse = client.put("/xdriver/test/value", "" + i).send().get();
                System.out.println(etcdKeysResponse);
            }
        }
        @Test
        public void testListener() throws Exception{
            this.startListenerThread(client , "/xdriver/test/value");
            Thread.sleep(1000000L);
        }
        public void startListenerThread(EtcdClient client , String dir) throws Exception{
            EtcdKeysResponse etcdKeysResponse = client.get(dir).send().get();
            System.out.println(etcdKeysResponse.getNode().getValue());
            new Thread(()->{
                startListener(client,dir,etcdKeysResponse.getNode().getModifiedIndex() + 1);
            }).start();
        }
        public void startListener(final EtcdClient client , final String dir , long waitIndex)  {
            EtcdResponsePromise<EtcdKeysResponse> promise = null;
            try {
                // 如果监听的waitIndex 不存在与key对应的EventHistroy 中（currentIndex >= waitIndex） ，
                // 并且key对应的modifyIndex > waitIndex , 则会查找到第一个大于waitIndex 的数据进行展示
                promise = client.get(dir).waitForChange(waitIndex).consistent().send();
            } catch (IOException e) {
                e.printStackTrace();
            }
            promise.addListener(promisea -> {
                try {
                    EtcdKeysResponse etcdKeysResponse = promisea.get();
                    new Thread(() -> {startListener(client , dir , etcdKeysResponse.getNode().getModifiedIndex() + 1);}).start();
                    String config = getConfig(dir, etcdKeysResponse);//加载配置项
                    System.out.println(config);
                } catch (Exception e) {
                    e.printStackTrace();
                    System.out.println("listen etcd 's config change cause exception:{}"+e.getMessage());
                }
            });
        }
    }

转载于:https://my.oschina.net/LucasZhu/blog/2875221

[2019_Python_]: https://my.oschina.net/u/2663968/blog/3061697
[hot3.png]: /images/20220114/d7bf7c0f88714ba982ba24651ecd1d91.png