为什么 Apache Curator 不触发所有更新？

Question

请在创建空白/test/a 路径后对您的 Zookeeper 服务器运行以下命令。

import static java.lang.String.valueOf;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.RetryForever;

public class CacheUpdateTest {
    static final String connectString = "127.0.0.1:2181,127.0.0.1:2191,127.0.0.1:2201";
    static volatile boolean stop = false;

    public static void main(String[] args) throws Exception {
        new Listener().start();
        Thread.sleep(1000);
        new Updater().start();
    }

    private static class Listener extends Thread {
        @SuppressWarnings("resource")
        @Override
        public void run() {
            CuratorFramework client = CuratorFrameworkFactory.builder().connectString(connectString).retryPolicy(new RetryForever(100)).build();
            client.start();

            PathChildrenCache cache = new PathChildrenCache(client, "/test", true);
            cache.getListenable().addListener(new PathChildrenCacheListener() {

                @Override
                public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                    if (event.getData() == null || event.getData().getData() == null) return;
                    int newI = Integer.parseInt(new String(event.getData().getData()));
                    System.err.println("Sensed update: " + newI);
                }
            });
            try {
                cache.start(StartMode.BUILD_INITIAL_CACHE);
                Thread.sleep(2000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private static class Updater extends Thread {
        @Override
        public void run() {
            try {
                CuratorFramework client = CuratorFrameworkFactory.builder().connectString(connectString).retryPolicy(new RetryForever(100)).build();
                client.start();

                for (int i = 0; i < 10; i++) {
                    // Thread.sleep(100);
                    System.out.println("Updated child: " + i);
                    client.setData().forPath("/test/a", valueOf(i).getBytes());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

如果我取消注释 Thread.sleep(100) 行，我通常会得到以下输出

Updated child: 0
Sensed update: 0
Updated child: 1
Sensed update: 1
Updated child: 2
Sensed update: 2
Updated child: 3
Sensed update: 3
Updated child: 4
Sensed update: 4
Updated child: 5
Sensed update: 5
Updated child: 6
Sensed update: 6
Updated child: 7
Sensed update: 7
Updated child: 8
Sensed update: 8
Updated child: 9
Sensed update: 9

当我评论它时，我得到以下输出

Updated child: 0
Updated child: 1
Sensed update: 1 --> Missed 0
Updated child: 2
Updated child: 3
Updated child: 4
Sensed update: 3 --> Missed 2
Updated child: 5
Updated child: 6
Sensed update: 5 --> Missed 4
Updated child: 7
Updated child: 8
Sensed update: 7 --> Missed 6
Updated child: 9
Sensed update: 9 --> Missed 8

为什么我并不总是收到所有通知？为什么我没有错过第一个？

Answer 1

Curator 是一个库，旨在简化 Apache Zookeeper 的使用。 PathChildrenCache 的工作方式是使用 ZK Watchers。

Watcher 创建一次性 Watch。如果 Watcher 收到有关更改（或它订阅的任何其他操作）的通知，则 Watcher 将被 Watcher 使用，并且它必须再次创建一个新 Watch 以在将来继续收到通知。

在您的情况下，PathChildrenCache 正在寻找节点中的更改。它的工作方式是等到它收到来自 ZK 的通知并重新创建 Watch 以继续寻找进一步的更改。

由于一切都是异步的，因此在您收到更改通知之前，数据可能已经更改了很多次。这就是为什么当您在更新程序中设置延迟时，您可以看到所有更改，因为缓存有足够的时间来检测更改并在调用新的setData 之前重新创建 Watch。当您忽略睡眠时，事情会发生得如此之快，以至于缓存会错过一些事件。

更多阅读请查看官方documentation关于watchers，主要是这个部分：

因为 watch 是一次性触发器，并且在获取事件和发送新请求以获取 watch 之间存在延迟，因此您无法可靠地看到 ZooKeeper 中节点发生的每一次更改。准备好处理 znode 在获取事件和再次设置手表之间多次更改的情况。 （你可能不在乎，但至少意识到它可能会发生。）