深入理解 ZooKeeper客户端与服务端的watcher回调（一）

public class ZookepperClientTest {
    public static void main(String[] args) throws Exception {
        ZooKeeper client = new ZooKeeper("localhost", 5000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.err.println("连接,触发");
            }
        });
    Stat stat = new Stat();
     //   todo 下面添加的事件监听器可是实现事件的消费订阅
      String content = new String(client.getData("/node1", new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                // todo 任何连接上这个节点的客户端修改了这个节点的 data数据,都会引起process函数的回调
                // todo 特点1:  watch只能使用1次
                if (event.getType().equals(Event.EventType.NodeDataChanged)){
                    System.err.println("当前节点数据发生了改变");
                }
            }
        }, stat));

public byte[] getData(final String path, Watcher watcher, Stat stat)
        throws KeeperException, InterruptedException
     {
         // todo 校验path
        final String clientPath = path;
        PathUtils.validatePath(clientPath);
        // the watch contains the un-chroot path
        WatchRegistration wcb = null;
        if (watcher != null) {
            // todo DataWatchRegistration 继承了 WatchRegistration
            // todo DataWatchRegistration 其实就是一个简单的内部类,将path 和 watch 封装进了一个对象
            wcb = new DataWatchRegistration(watcher, clientPath);
        }
        final String serverPath = prependChroot(clientPath);
        // todo 创建一个请求头
        RequestHeader h = new RequestHeader();
        h.setType(ZooDefs.OpCode.getData);
        // todo 创建了一个GetDataRequest
        GetDataRequest request = new GetDataRequest();
        // todo 给这个请求初始化,path 是传递进来的path,但是 watcher不是!!! 如果我们给定了watcher , 这里面的条件就是  true
        request.setPath(serverPath);
        request.setWatch(watcher != null); // todo 可看看看服务端接收到请求是怎么办的
        GetDataResponse response = new GetDataResponse();
        // todo 同样由 clientCnxn 上下文进行提交请求, 这个操作应该同样是阻塞的
         // todo EventThread 和 SendThread 同时使用一份 clientCnxn的 submitRequest()
        ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
        if (r.getErr() != 0) {
            throw KeeperException.create(KeeperException.Code.get(r.getErr()),
                    clientPath);
        }
        if (stat != null) {
            DataTree.copyStat(response.getStat(), stat);
        }
        return response.getData();
    }

public ReplyHeader submitRequest(RequestHeader h, Record request, Record response, WatchRegistration watchRegistration)
        throws InterruptedException {
    ReplyHeader r = new ReplyHeader();
    // todo 来到这个 queuePacket() 方法在下面, 这个方法就是将  用户输入-> string ->>> request ->>> packet 的过程
    Packet packet = queuePacket(h, r, request, response, null, null, null,
            null, watchRegistration);
    // todo 使用同步代码块,在下面的进行    同步阻塞等待, 直到有了Response响应才会跳出这个循环, 这个finished状态就是在客户端接受到服务端的
    // todo 的响应后, 将服务端的响应解析出来,然后放置到 pendingqueue里时,设置上去的
    synchronized (packet) {
        while (!packet.finished) {
            // todo 这个等待是需要唤醒的
            packet.wait();
        }
    }
    // todo 直到上面的代码块被唤醒,才会这个方法才会返回
    return r;
}

case OpCode.getData: {
        lastOp = "GETD";
        GetDataRequest getDataRequest = new GetDataRequest();
        ByteBufferInputStream.byteBuffer2Record(request.request,
                getDataRequest);
        DataNode n = zks.getZKDatabase().getNode(getDataRequest.getPath());
        if (n == null) {
            throw new KeeperException.NoNodeException();
        }
        PrepRequestProcessor.checkACL(zks, zks.getZKDatabase().aclForNode(n),
                ZooDefs.Perms.READ,
                request.authInfo);
        Stat stat = new Stat();
        // todo 这里的操作    getDataRequest.getWatch() ? cnxn : null 对应可客户端的  跟进watcher有没有而决定往服务端传递 true 还是false 相关
        // todo 跟进去 getData()
        byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat,
                getDataRequest.getWatch() ? cnxn : null);
        //todo  cnxn的Processor()被回调, 往客户端发送数据 , 什么时候触发呢? 就是上面的  处理事务时的回调 第127行
        // todo 构建了一个 rsp ,在本类的最后面将rsp 响应给client
        rsp = new GetDataResponse(b, stat);
        break;
    }

public byte[] getData(String path, Stat stat, Watcher watcher)
        throws KeeperException.NoNodeException {
    DataNode n = nodes.get(path);
    if (n == null) {
        throw new KeeperException.NoNodeException();
    }
    synchronized (n) {
        n.copyStat(stat);
        if (watcher != null) {
            // todo 将path 和 watcher 绑定在一起
            dataWatches.addWatch(path, watcher);
        }
        return n.data;
    }
}

// todo start就是启动了在构造方法中创建的线程
    public void start() {
        sendThread.start();
        eventThread.start();
    }

// todo zookeeper客户端, 处理用户输入命令的具体逻辑
    // todo  用大白话讲,下面其实就是把 从控制台获取的用户的输入信息转换成指定的字符, 然后发送到服务端
    // todo MyCommandOptions 是处理命令行选项和shell脚本的工具类
    protected boolean processZKCmd(MyCommandOptions co) throws KeeperException, IOException, InterruptedException {
        // todo 在这个方法中可以看到很多的命令行所支持的命令
        Stat stat = new Stat();
        // todo 获取命令行输入中 0 1 2 3 ... 位置的内容, 比如 0 位置是命令  1 2 3 位置可能就是不同的参数
        String[] args = co.getArgArray();
        String cmd = co.getCommand();
        if (args.length < 1) {
            usage();
            return false;
        }
        if (!commandMap.containsKey(cmd)) {
            usage();
            return false;
        }
        boolean watch = args.length > 2;
        String path = null;
        List<ACL> acl = Ids.OPEN_ACL_UNSAFE;
        LOG.debug("Processing " + cmd);
        if (cmd.equals("quit")) {
            System.out.println("Quitting...");
            zk.close();
            System.exit(0);
        } else if (cmd.equals("set") && args.length >= 3) {
            path = args[1];
            stat = zk.setData(path, args[2].getBytes(),
                    args.length > 3 ? Integer.parseInt(args[3]) : -1);
            printStat(stat);

public Stat setData(final String path, byte data[], int version)
        throws KeeperException, InterruptedException
    {
        final String clientPath = path;
        PathUtils.validatePath(clientPath);
        final String serverPath = prependChroot(clientPath);
        RequestHeader h = new RequestHeader();
        h.setType(ZooDefs.OpCode.setData);
        SetDataRequest request = new SetDataRequest();
        request.setPath(serverPath);
        request.setData(data);
        request.setVersion(version);
        SetDataResponse response = new SetDataResponse();
        ReplyHeader r = cnxn.submitRequest(h, request, response, null);
        if (r.getErr() != 0) {
            throw KeeperException.create(KeeperException.Code.get(r.getErr()),
                    clientPath);
        }
        return response.getStat();
    }