zookeeper 快速入门

zookeeper

zookeeper是什么

Apache ZooKeeper是Apache软件基金会的一个软件项目，他为大型分布式计算提供开源的分布式配置服务、同步服务和命名注册。ZooKeeper曾经是Hadoop的一个子项目，但如今是一个独立的顶级项目。

ZooKeeper的架构经过冗余服务实现高可用性。所以，若是第一次无应答，客户端就能够询问另外一台ZooKeeper主机。ZooKeeper节点将它们的数据存储于一个分层的命名空间，很是相似于一个文件系统或一个前缀树结构。客户端能够在节点读写，从而以这种方式拥有一个共享的配置服务。

使用ZooKeeper的公司包括Rackspace、雅虎和eBay，以及相似于像Solr这样的开源企业级搜索系统。

zookeeper提供了什么

• 文件系统：zookeeper维护一个相似文件系统的数据结构，每一个子目录项如 NameService 都被称做为 znode，和文件系统同样，自由增长及删除，惟一不一样其可存储数据。Znode分为四种类型

  • PERSISTENT-持久化目录节点。（客户端与zookeeper断开链接后，该节点依旧存在）。
  • PERSISTENT_SEQUENTIAL-持久化顺序编号目录节点。（客户端与zookeeper断开链接后，该节点依旧存在，只是Zookeeper给该节点名称进行顺序编号）
  • EPHEMERAL-临时目录节点（客户端与zookeeper断开链接后，该节点被删除）
  • EPHEMERAL_SEQUENTIAL-临时顺序编号目录节点。（客户端与zookeeper断开链接后，该节点被删除，只是Zookeeper给该节点名称进行顺序编号）
• 通知机制：客户端注册监听它关心的目录节点，当目录节点发生变化（数据改变、被删除、子目录节点增长删除）时，zookeeper会通知客户端。

zookeeper能为咱们作什么？

• 命名服务：在zookeeper的文件系统里建立一个目录，即有惟一的path。在咱们使用tborg没法肯定上游程序的部署机器时便可与下游程序约定好path，经过path即能互相探索发现。
• 配置管理：把应用配置放置zookeeper上去,保存在 Zookeeper 的某个目录节点中，而后全部相关应用程序对这个目录节点进行监听，一旦配置信息发生变化，每一个应用程序就会收到 Zookeeper 的通知，而后从 Zookeeper 获取新的配置信息应用到系统中就好。
• 集群管理：节点（机器）增删及Master选取。节点增删：全部机器约定在父目录GroupMembers下建立临时目录节点，而后监听父目录节点的子节点变化消息。一旦有机器挂掉，该机器与 zookeeper的链接断开，其所建立的临时目录节点被删除，全部其余机器都收到通知：某个兄弟目录被删除，因而，全部人都知道：它上船了。新机器加入 也是相似，全部机器收到通知：新兄弟目录加入，highcount又有了。Master选取：全部机器建立临时顺序编号目录节点，每次选取编号最小的机器做为master就好。
• 分布式锁：基于zookeeper一致性文件系统,实现锁服务。锁服务分为保存独占及时序控制两类。保存独占：将zookeeper上的一个znode看做是一把锁，经过createznode的方式来实现。全部客户端都去建立 /distribute_lock 节点，最终成功建立的那个客户端也即拥有了这把锁。用完删除本身建立的distribute_lock 节点就释放锁。时序控制：基于/distribute_lock锁，全部客户端在它下面建立临时顺序编号目录节点，和选master同样，编号最小的得到锁，用完删除，依次方便。
• 队列管理：分同步队列,FIFO队列（入队与出队），同步队列：当一个队列的成员都聚齐时，这个队列才可用，不然一直等待全部成员到达。在约定目录下建立临时目录节点，监听节点数目是不是咱们要求的数目。FIFO队列：和分布式锁服务中的控制时序场景基本原理一致，入列有编号，出列按编号。
• 分布式与数据复制：Zookeeper做为一个集群提供一致的数据服务，必然在全部机器间作数据复制。数据复制好处：（1）容错：一个节点出错，不致于让整个系统中止工做，别的节点能够接管它的工做。（2）提升系统的扩展能力：把负载分布到多个节点上，或者增长节点来提升系统的负载能力；（3）性能提高：让客户端本地访问就近节点,提升用户访问速度。

zookeeper基本概念

角色简介

Zookeeper角色分为三类，领导者：负责进行投票的发起和决议,更新系统状态。跟随者：Follower用于接收客户请求并向客户端返回结果，在选中过程当中参与投票。观察者：Observer能够接收客户端链接，将写请求转发给leader节点。但不参加投票过程,只同步leader状态。Observer目的在于扩展系统，提升读取速度。

设计目的

• 一致性：client不论链接到哪一个Server，展现给它都是同一个视图，这是zookeeper最重要的性能。
• 可靠性：具备简单、健壮、良好的性能，若是消息m被到一台服务器接受，那么它将被全部的服务器接受。
• 实时性：Zookeeper保证客户端将在一个时间间隔范围内得到服务器的更新信息，或者服务器失效的信息。但因为网络延时等缘由，Zookeeper不能保证两个客户端能同时获得刚更新的数据，若是须要最新数据，应该在读数据以前调用sync()接口。
• 等待无关（wait-free）：慢的或者失效的client不得干预快速的client的请求，使得每一个client都能有效的等待。
• 原子性：更新只能成功或者失败，没有中间状态。
• 顺序性：包括全局有序和偏序两种：全局有序是指若是在一台服务器上消息a在消息b前发布，则在全部Server上消息a都将在消息b前被发布；偏序是指若是一个消息b在消息a后被同一个发送者发布，a必将排在b前面。

选主流程

当leader崩溃或者leader失去大多数的follower，这时候zk进入恢复模式，恢复模式须要从新选举出一个新的leader，让全部的Server都恢复到一个正确的状态。Zk的选举算法有两种：一种是基于basic paxos实现的，另一种是基于fast paxos算法实现的。系统默认的选举算法为fast paxos。先介绍basic paxos流程：

选举线程由当前Server发起选举的线程担任，其主要功能是对投票结果进行统计，并选出推荐的Server；
选举线程首先向全部Server发起一次询问(包括本身)；
选举线程收到回复后，验证是不是本身发起的询问(验证zxid是否一致)，而后获取对方的id(myid)，并存储到当前询问对象列表中，最后获取对方提议的leader相关信息(id,zxid)，并将这些信息存储到当次选举的投票记录表中；
收到全部Server回复之后，就计算出zxid最大的那个Server，并将这个Server相关信息设置成下一次要投票的Server；
线程将当前zxid最大的Server设置为当前Server要推荐的Leader，若是此时获胜的Server得到n/2 + 1的Server票数， 设置当前推荐的leader为获胜的Server，将根据获胜的Server相关信息设置本身的状态，不然，继续这个过程，直到leader被选举出来。
经过流程分析咱们能够得出：要使Leader得到多数Server的支持，则Server总数必须是奇数2n+1，且存活的Server的数目不得少于n+1.

每一个Server启动后都会重复以上流程。在恢复模式下，若是是刚从崩溃状态恢复的或者刚启动的server还会从磁盘快照中恢复数据和会话信息，zk会记录事务日志并按期进行快照，方便在恢复时进行状态恢复。

fast paxos流程是在选举过程当中，某Server首先向全部Server提议本身要成为leader，当其它Server收到提议之后，解决epoch和zxid的冲突，并接受对方的提议，而后向对方发送接受提议完成的消息，重复这个流程，最后必定能选举出Leader。

zookeeper的安装使用

wget http://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.8/zookeeper-3.4.8.tar.gz

tar zxvf zookeeper-3.4.8.tar.gz -C /usr/local/

cd $ZOOKEEPER_HOME

cp conf/zoo_sample.cfg conf/zoo.cfg

# 集群须要在zoo.cfg配置

server.1=192.168.1.148:2888:3888
server.2=192.168.1.149:2888:3888
server.3=192.168.1.150:2888:3888


# 在zookeeper的临时目录建立myid
mkdir -p /tmp/zookeeper

# 分别在不一样节点建立myid文件里面的数字对应节点的编号好比server.1就对应1，server.2就对应2
echo 1 > /tmp/zookeeper/myid

# 最后分别启动集群上的节点
$ZOOKEEPER_HOME/bin/zkServer.sh start

# 查看zookeeper的状态
$ZOOKEEPER_HOME/bin/zkServer.sh status

# 中止zookeeper服务
$ZOOKEEPER_HOME/bin/zkServer.sh stop

zookeeper命令行操做

启动zookeeper服务后到bin目录启动zookeeper的客户端$ZOOKEEPER_HOME/bin/zkCli.sh

# 输入help

[zk: localhost:2181(CONNECTED) 0] help
ZooKeeper -server host:port cmd args
        stat path [watch]
        set path data [version]
        ls path [watch]
        delquota [-n|-b] path
        ls2 path [watch]
        setAcl path acl
        setquota -n|-b val path
        history 
        redo cmdno
        printwatches on|off
        delete path [version]
        sync path
        listquota path
        rmr path
        get path [watch]
        create [-s] [-e] path data acl
        addauth scheme auth
        quit 
        getAcl path
        close 
        connect host:port

建立节点

[zk: localhost:2181(CONNECTED) 14] create /test test-data
Created /test

查看节点

[zk: localhost:2181(CONNECTED) 1] ls /
[abc, zookeeper, eclipse]

获取节点

[zk: localhost:2181(CONNECTED) 10] get /test
test-update
cZxid = 0x38
ctime = Sat Dec 22 10:11:46 CST 2018
mZxid = 0x39
mtime = Sat Dec 22 10:12:05 CST 2018
pZxid = 0x38
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 11
numChildren = 0

修改节点

[zk: localhost:2181(CONNECTED) 11] set /test test-update
cZxid = 0x38
ctime = Sat Dec 22 10:11:46 CST 2018
mZxid = 0x3a
mtime = Sat Dec 22 10:15:04 CST 2018
pZxid = 0x38
cversion = 0
dataVersion = 2
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 11
numChildren = 0

删除节点

[zk: localhost:2181(CONNECTED) 13] delete /test

zookeeper java 客户端操做

/**
 * @author leone
 * @since 2018-06-16
 **/
public class ZkClient {

    private final static Logger logger = LoggerFactory.getLogger(ZkClient.class);

    private final static String ZK_URL = "xxx.xxx.xxx.xxx:2181";

    private final static int TIME_OUT = 5000;

    private static ZooKeeper zkClient = null;


    @Before
    public void init() throws Exception {
        zkClient = new ZooKeeper(ZK_URL, TIME_OUT, (WatchedEvent event) -> {
            // 收到事件通知后的回调函数（应该是咱们本身的事件处理逻辑）
            logger.info(event.getType() + "---" + event.getPath());
            try {
                zkClient.getChildren("/", true);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    /**
     * 设置值
     *
     * @throws Exception
     */
    @Test
    public void testSetData() throws Exception {
        zkClient.setData("/eclipse", "world".getBytes(), -1);
        byte[] data = zkClient.getData("/eclipse", false, null);
        System.out.println(new String(data));
    }

    /**
     * 建立节点
     *
     * @throws Exception
     */
    @Test
    public void testCreate() throws Exception {
        // 参数1：要建立的节点的路径 参数2：节点数据 参数3：节点的权限 参数4：节点的类型
        zkClient.create("/eclipse/aaa", "aaaData".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }


    /**
     * 测试某节点是否存在
     *
     * @throws Exception
     */
    @Test
    public void testExists() throws Exception {
        Stat stat = zkClient.exists("/eclipse", false);
        System.out.println(stat == null ? "not exist" : "exist");
    }

    /**
     * 获取子节点
     *
     * @throws Exception
     */
    @Test
    public void testGetChild() throws Exception {
        List<String> children = zkClient.getChildren("/", true);
        for (String child : children) {
            System.out.println(child);
        }
    }

    /**
     * 删除节点
     *
     * @throws Exception
     */
    @Test
    public void testDelete() throws Exception {
        // 参数2：指定要删除的版本，-1表示删除全部版本
        zkClient.delete("/abc", -1);
    }


    /**
     * 获取节点的数据
     *
     * @throws Exception
     */
    @Test
    public void testGetDate() throws Exception {
        byte[] data = zkClient.getData("/eclipse", false, null);
        System.out.println(new String(data));
    }

}