etcd事件监听

    etcd是CoreOS开发的一个高可用的键值存储系统，主要用于共享配置和服务发现。它使用Go语言编写，并通过Raft一致性算法处理日志复制以保证强一致性。etcd目前的最新版本是2.2.0。

    和zookeeper的二进制接口不同，它提供了HTTP/JSON的rest api接口，所以对使用它的客户端来说是很友好的，几乎每种编程语言都有较成熟的http client开发包，基于这些开发包很容易编写etcd客户端。在java中已经有个比较好的客户端etcd4j(https://github.com/jurmous/etcd4j)，类似zookeeper有好用的curator库。

   

   etcd事件

    etcd中和数据变化（包括目录&值key、value变化）相关的事件类型有：set, delete, update, create, compareAndSwap、compareAndDelete。在etcd http response (json格式）action属性就是事件类型。

     etcd的事件watch监听接口也是使用http访问，它提供两种监听模式，一种是一次性监听，类似zookeeper的事件watch，监听到一次事件后，需要客户端重新发起监听，比较繁琐。另一种是持久监听（stream），当有事件时，会连续触发，不需要客户端重新发起监听。

     和zookeeper一样也存在客户端丢失监听事件的可能，它不保证每个事件客户端都能监听到，不过我们在实际使用过程中，通常是在应用启动时，在开始监听之后或之前先查询etcd，将需要的数据全量加载到内存中，后续才是根据监听事件来增量更新内存中的数据或全量刷新数据。另外目前版本的etcd最多只保存1千条事件历史，满1千条后，如果有新的事件产生，事件历史库中最老的事件将被丟弃，这个可以理解为etcd服务端事件丢失。由于有1千条的限制，在有事件浪涌时（在单位时间内例如1秒内产生2千条事件），事件监听处理较慢的时候或未监听时也会发生客户端事件丢失。   

 

 

    一次性watch监听

   可用curl发送http请求来进行演示，假设在本机安装有etcd，客户端访问端口为2379(旧版本监听端口为4001)，监听key为/application。一次性监听命令如下：

    curl "http://127.0.0.1:2379/v2/keys/application?wait=true"

    查询字符串中的wait=true表示监听，还可加查询参数recursive=true表示附加监听/application的子目录和后辈目录，例如curl "http://127.0.0.1:2379/v2/keys/application?wait=true&recursive=true" 。

    启动监听后etcd如有变化事件发生，则返回json格式的事件，结果类似如下：

    HTTP/1.1 200 OK

    Content-Type: application/json

    X-Etcd-Cluster-Id: e88d54f6225f06ad

    X-Etcd-Index: 271

    X-Raft-Index: 872202

    X-Raft-Term: 5

    Date: Sat, 26 Sep 2015 08:43:17 GMT

    Transfer-Encoding: chunked

 

    {"action":"set","node":{"key":"/application/store","value":"        {...}","modifiedIndex":271,"createdIndex":271},"prevNode":{...}}

   上面的curl返回结果中显示了http头，是加了curl的-i选项。

   从结果可看出action属性为set事件，modifiedIndex是该事件的index，为271。X-Etcd-Index头表示启动监听时etcd的当前index。etcd的index是单调递增的正整数，该整数取值空间是etcd中的所有key共享的。

    由于是一次性监听，所以curl会退出，继续监听要重新运行curl。

    和zookeeper事件监听一样，上面的http请求只能监听到命令发出之后产生的事件(更精确的说是在etcd服务器收到监听请求之后)，之前发生的事件是监听不到的。另外在前一个监听得到触发开始(中间包括事件处理耗时)到启动下一次监听之间(并且下一次监听是etcd服务器收到监听请求才真正开始生效，这中间还有网络时延)肯定存在时间间隔，如果在这个时间间隔（区间）内有事件发生，客户端是监听不到这些事件的，这个也和zookeeper类似。也就是如果两个事件并行发生或在发生时间上相距很近（例如update一个key后立即delete），后一事件的发生刚好落在这个间隔内，那这个事件在采用一次性监听模式的客户端中会丢失，这些事件对客户端监听不可见。

    

 

    那是不是就一定不能获取先前的事件了？这个要看情况，只要事件还在etcd的事件历史中，并且知道事件的index或起始index，还是可以取到单个事件的。在查询字符串中加waitIndex参数，其值是该事件的index或起始index。例如curl "http://127.0.0.1:2379/v2/keys/application?wait=true&recursive=true&waitIndex=269" ，表示查询/application目录或后辈目录中index等于或大于269的事件，如果有则返回一个最接近269的事件，否则会挂起curl，直到有满足条件的事件发生。这个其实是查询事件，已经不是正规的事件监听了。所以如果知道起始index，可以每次index加1来遍历查询事件，遍历查询时可以根据返回事件中的index来调整后续查询条件中使用的index。 

    如果waitIndex对应的事件已被etcd丢弃（见前面提到的etcd事件历史库），此时如果获取该事件，etcd将返回http状态码400( Bad Request)，http body类似如下：

    {"errorCode":401,"message":"The event in requested index is outdated and cleared","cause":"the requested history has been cleared [3305/1]","index":4304} 

    另外可以监听根目录（根key），例如curl "http://127.0.0.1:2379/v2/keys?wait=true&recursive=true" 。

 

    持久监听(流式watch)

    要在查询字符串中加stream=true，例如curl "http://127.0.0.1:2379/v2/keys/application?wait=true&recursive=true&stream=true"

    发出上述命令后，会和etcd服务器之间建立一个http长连接（此处的长连接不是指http keep alive，虽然也需要keep alive），其应答http头中的Transfer-Encoding为chunked，类似server push和comet中的multipart/x-mixed-replace，其http response body不结束，后续的每个事件作为http body的一部分，在该长连接中以一个或多个body chunk块的形式推送给客户端。可以看到在流式持久监听中curl即使收到事件也不会退出，它一直在等待后续将要发生的事件。一次性监听http头中也是chunked，但收到一个事件后，etcd会发送一个结束chunk（大小为0，表示http response结束），因此收到该chunk后curl会退出。

     相比一次性监听，除了更简便，持久监听也比一次性监听可靠性高，不会出现上面提到的在时间间隔内监听不到事件的情况。

      持久监听也是监听从命令发出之后发生的事件，先前的事件是监听不到的，不像JMS的持久订阅可以收到启动之前的jms消息。

      如果在持久监听中加waitIndex参数，分两种情况：一种是waitIndex的值小于或等于启动监听时etcd的当前index（这个值在X-Etcd-Index http头中可看到），则curl接收到满足条件的事件后挂起（不退出），后续再也收不到其他事件，即最多只能收到一个事件。第二种是waitIndex的值大于启动监听时etcd的当前index，则waitIndex参数无效，持久监听的行为同没有waitIndex参数一样。

 

 

   使用etcd4j进行事件监听

   在使用etcd4j进行事件监听时，有个注意事项:如果在同一个jvm虚拟机中既有修改查询etcd数据的操作，也有监听etcd事件，监听事件使用的EtcdClient实例和修改&查询操作使用的EtcdClient实例应分开，不能共用同一个实例，否则目前的etcd4j（2.7.0版本）会出现问题。多个key可以共用一个EtcdClient实例来进行监听，也可每个key使用一个单独的实例来进行监听。

• 一次性监听

  //创建etcd客户端实例

  EtcdClient   etcdClient = new EtcdClient (new URI[]{new URI("127.0.0.1:2379")});

  /*setRetryPolicy设置重试策略

  * recursive()监听/application目录和其后辈目录中的事件，设置查询字符串recursive=true.

  *  waitForChange()监听，设置查询字符串wait=true.

 * -1表示永不超时，send()发送http请求进行监听

  */

  EtcdResponsePromise<EtcdKeysResponse> promise = etcd.get("/application").setRetryPolicy(

                        new RetryNTimes(300,Integer.MAX_VALUE)).recursive().waitForChange().

                        timeout(-1, TimeUnit.SECONDS).send();

    //增加listener

    promise.addListener(new IsSimplePromiseResponseHandler<EtcdKeysResponse>()

    {

         @Override

         public void onResponse(ResponsePromise<EtcdKeysResponse> promise) {

                   //从promise中取响应，检查是否有异常

                   EtcdKeysResponse response = promise.getNow();

                   Throwable  exception = promise.getException();

                    ... 处理事件

                    //再次发起http请求进行监听

                    ...

         }   

     }

   );

 

     还可使用waitForChange(long waitIndex)来监听指定的事件。

 

• 持久监听

     etcd4j的当前版本2.7.0不支持持久监听，可以使用

Java异步async httpclient（https://github.com/AsyncHttpClient/async-http-client）来实现持久监听，最挫最原始的持久监听实验代码如下：

     AsyncHttpClient asyncHttpClient = ...;

      //发送http get异步请求，不超时，且提供一个匿名的异步请求完成handler

      asyncHttpClient.prepareGet("http://127.0.0.1:2379/v2/keys/application?   wait=true&recursive=true&stream=true").setRequestTimeout(-1).execute(

       new AsyncHandler<String>() {

           //这个异常回调方法可类比为zookeeper的会话过期

           @Override

           public void onThrowable(Throwable t) {

                  // TODO Auto-generated method stub

                  //todo:异常，重新重试发起监听

            }

             

             //在会话期，有事件时会回调onBodyPartReceived方法

           @Override

           public State  onBodyPartReceived(HttpResponseBodyPart  bodyPart)

throws Exception {

                 // TODO Auto-generated method stub

                // bodyPart对应一个http chunk块，一个事件的数据可能由多个chunk块组成

                System.out.println("event="+new String(bodyPart.getBodyByteBuffer().array(),"utf-8"));

                //todo:将chunk中的数据加入缓冲区，待收到完整的事件数据后将json格式

                //的数据解析成java对象

               ...

               return State.CONTINUE;

         }

  

        @Override

        public  State onStatusReceived(HttpResponseStatus  responseStatus)

throws Exception {

               // TODO Auto-generated method stub

               return State.CONTINUE;

         }

            

            //一个会话期只调用一次，这个回调方法可类比为zookeeper会话建立

         @Override

         public  State onHeadersReceived(HttpResponseHeaders headers)

throws Exception {

               // TODO Auto-generated method stub

               return State.CONTINUE;

          }

 

         @Override

          public String onCompleted() throws Exception {

                    // TODO Auto-generated method stub

                   return "endWatch";

          }

   

    });