Zookeeper 概念 - 爱码网

Zookeeper:

Zookeeper是一个高可用的分布式管理与协调框架，基于ZAB算法（原子消息广播协议）的实现。该框架能够很好的保证分布式环境中数据的一致性。也只是基于这样的特性，使得Zookeeper成为了解决分布式一致性问题的利器。

Zookeeper的特性：

①顺序一致性：从一个客户端发起的事务请求，最终将会严格的按照其发起的顺序被应用到zookeeper中去。

②原子性：所有事务请求的处理结果在整个集群中所有机器上的应用情况是一致的，也就是说，要么整个集群所有的机器都成功应用了某一事务，要么没有应用，一定不会出现部分机器应用了该事务，而另一部分没有应用的情况。

③单一视图：无论客户端连接的是哪一个zookeeper服务器，其看到的服务器端数据模型都是一致的。

④可靠性：一旦服务器成功地应用了一个事务，并完成对客户端的响应，那么该事务所引起的服务器端状态将会被一直保留，除非有另一个事务对其更改。

⑤实时性：通常所说的实时性就是指一旦事务被成功应用，那么客户端就能立刻从服务器上获取变更后的新数据，zookeeper仅仅能保证在一段时间内，客户端最终能从服务器端读取最新的数据状态。

Zookeeper原生api操作：

package com.qhong.zookeeper;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;

import java.util.concurrent.CountDownLatch;

public class ZookeeperBase {

    /** zookeeper地址 */
    //static final String CONNECT_ADDR = "192.168.80.88:2181,192.168.80.87:2181,192.168.80.86:2181";
    static final String CONNECT_ADDR="192.168.1.108:2181";
    /** session超时时间 */
    static final int SESSION_OUTTIME = 2000;//ms
    /** 信号量，阻塞程序执行，用于等待zookeeper连接成功，发送成功信号 */
    static final CountDownLatch connectedSemaphore = new CountDownLatch(1);

    public static void main(String[] args) throws Exception{

        ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, new Watcher(){
            @Override
            public void process(WatchedEvent event) {
                //获取事件的状态
                KeeperState keeperState = event.getState();
                EventType eventType = event.getType();
                //如果是建立连接
                if(KeeperState.SyncConnected == keeperState){
                    if(EventType.None == eventType){
                        //如果建立连接成功，则发送信号量，让后续阻塞程序向下执行
                        connectedSemaphore.countDown();
                        System.out.println("zk 建立连接");
                    }
                }
            }
        });

        //进行阻塞
        connectedSemaphore.await();

        System.out.println("..");
        //创建父节点
        zk.create("/testRoot", "testRoot".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        //创建子节点，使用EPHEMERAL，主程序执行完成后该节点被删除，只在本次会话内有效，可以用作分布式锁。
        zk.create("/testRoot/children", "children data".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

        //获取节点信息
        byte[] data = zk.getData("/testRoot", false, null);
        System.out.println(new String(data));
        System.out.println(zk.getChildren("/testRoot", false));

        //修改节点的值，-1表示跳过版本检查，其他正数表示如果传入的版本号与当前版本号不一致，则修改不成功，删除是同样的道理。
        zk.setData("/testRoot", "modify data root".getBytes(), -1);
        byte[] data2 = zk.getData("/testRoot", false, null);
        System.out.println(new String(data2));
        System.out.println("zk children is exists");
        //判断节点是否存在
        System.out.println(zk.exists("/testRoot/children", false));
        System.out.println("delete children");
        //删除节点
        zk.delete("/testRoot/children", -1);
        System.out.println(zk.exists("/testRoot/children", false));

        zk.close();
    }
}

Output:

zk 建立连接
..
testRoot
[children]
modify data root
zk children is exists
1309,1309,1511968640670,1511968640670,0,0,0,99087986199232525,13,0,1309

delete children
null

View Code