Watch 方法

WatchService 只有一个 Watch 方法。

  1. // Watch 观察将要发生或者已经发生的事件。
  2. // 输入和输出都是流;输入流用于创建和取消观察，而输出流发送事件。
  3. // 一个观察 RPC 可以在一次性在多个 key 范围上观察，并为多个观察流化事件。
  4. // 整个事件历史可以从最后压缩修订版本开始观察。
  5. rpc Watch(stream WatchRequest) returns (stream WatchResponse) {}

消息定义

请求的消息体是 WatchRequest：

  1. message WatchRequest {
  2.   // request_union 要么是创建新的观察者的请求，要么是取消一个已经存在的观察者的请求
  3.   oneof request_union {
  4.     WatchCreateRequest create_request = 1;
  5.     WatchCancelRequest cancel_request = 2;
  6.   }
  7. }

创建新的观察者的请求 WatchCreateRequest：

  1. message WatchCreateRequest {
  2.   // key 是注册要观察的 key
  3.   bytes key = 1;
  5.   // range_end 是要观察的范围 [key, range_end) 的终点。
  6.   // 如果 range_end 没有设置，则只有参数 key 被观察。
  7.   // 如果 range_end 等同于 '\0'， 则大于等于参数 key 的所有 key 都将被观察
  8.   // 如果 range_end 比给定 key 大1， 则所有以给定 key 为前缀的 key 都将被观察
  9.   bytes range_end = 2;
  11.   // start_revision 是可选的开始(包括)观察的修订版本。不设置 start_revision 则表示 "现在".
  12.   int64 start_revision = 3;
  14.   // 设置 progress_notify ，这样如果最近没有事件，etcd 服务器将定期的发送不带任何事件的 WatchResponse 给新的观察者。
  15.   // 当客户端希望从最近已知的修订版本开始恢复断开的观察者时有用。
  16.   // etcd 服务器将基于当前负载决定它发送通知的频率。
  17.   bool progress_notify = 4;
  19.   enum FilterType {
  20.   // 过滤掉 put 事件
  21.   NOPUT = 0;
  23.   // 过滤掉 delete 事件
  24.   NODELETE = 1;
  25.   }
  27.   // 过滤器，在服务器端发送事件给回观察者之前，过滤掉事件。
  28.   repeated FilterType filters = 5;
  30.   // 如果 prev_kv 被设置，被创建的观察者在事件发生前获取上一次的KV。
  31.   // 如果上一次的KV已经被压缩，则不会返回任何东西
  32.   bool prev_kv = 6;
  33. }

取消已有观察者的 WatchCancelRequest ：

  1. message WatchCancelRequest {
  2.   // watch_id 是要取消的观察者的id，这样就不再有更多事件传播过来了。
  3.   int64 watch_id = 1;
  4. }

应答的消息体 WatchResponse：

  1. message WatchResponse {
  2.   ResponseHeader header = 1;
  3.   // watch_id 是和应答相关的观察者的ID
  4.   int64 watch_id = 2;
  6.   // 如果应答是用于创建观察者请求的，则 created 设置为 true。
  7.   // 客户端应该记录 watch_id 并期待从同样的流中为创建的观察者接收事件。
  8.   // 所有发送给被创建的观察者的事件将附带同样的 watch_id
  9.   bool created = 3;
  11.   // 如果应答是用于取消观察者请求的，则 canceled 设置为true。
  12.   // 不会再有事件发送给被取消的观察者。
  13.   bool canceled = 4;
  15.   // compact_revision 被设置为最小 index，如果观察者试图观察被压缩的 index。
  16.   // 当在被压缩的修订版本上创建观察者或者观察者无法追上键值对存储的进展时发生。
  17.   // 客户端应该视观察者为被取消，并不应该试图再次创建任何带有相同 start_revision 的观察者。
  18.   int64 compact_revision  = 5;
  20.   // cancel_reason 指出取消观察者的理由.
  21.   string cancel_reason = 6;
  23.   repeated mvccpb.Event events = 11;
  24. }

mvccpb.Event 的消息体：

  1. message Event {
  2.   enum EventType {
  3.     PUT = 0;
  4.     DELETE = 1;
  5.   }
  7.   // type 是事件的类型。
  8.   // 如果类型是 PUT，表明新的数据已经存储到 key。
  9.   // 如果类型是 DELETE， 表明 key 已经被删除。
  10.   EventType type = 1;
  12.   // kv 为事件持有 KeyValue。
  13.   // PUT 事件包含当前的kv键值对
  14.   // kv.Version=1 的 PUT 事件表明 key 的创建
  15.   // DELETE/EXPIRE 事件包含被删除的 key，它的修改修订版本设置为删除的修订版本
  16.   KeyValue kv = 2;
  18.   // prev_kv 持有在事件发生前的键值对
  19.   KeyValue prev_kv = 3;
  20. }