大数据学习教程SD版第六篇【Zookeeper】

大数据学习教程SD版第六篇【Zookeeper】 1. Zookeeper 简介

开源，分布式，为分布式框架提供协调服务，等价于：文件系统+通知机制

• 工作机制

1. 服务端节点启动时，向ZK注册信息【临时节点】
2. Client 获取当前服务器列表，并注册监听
3. 当服务器节点下线
4. ZK 通知Client
5. Client 重新获取服务器列表，并注册监听

• 特点

1. 主从架构，一个Leader,多个Follower
2. 半数以上节点存活，ZK 集群正常工作
3. 全局数据一致，每个Server存一份相同数据副本
4. Client更新请求，顺序执行
5. 数据更新具有原子性
6. 实时性，最新数据

• 数据结构：类似Unix文件系统

1. 一个ZNode 默认 存储 1MB 数据
2. 每个ZNode 都有唯一标识

• 功能场景

1. 统一命令服务
2. 统一配置管理
3. 统一集群管理
4. 服务器动态上下线
5. 软负载均衡

2. Zookeeper 安装 2.1 本地安装

环境准备：JDK

1. 下载安装包
2. 解压安装包
3. 修改配置文件 zoo.cfg

# 集群通信心跳时间ms
tickTime=2000
# Leader 和 Follow 初次初始化时间 10*tickTime
initLimit=10
# Leader 和 Follow 同步时间 5*tickTime
syncLimit=5
# Zk 数据存放目录
dataDir=/opt/module/apache-zookeeper-3.5.7-bin/zkData
# Zk 通信端口
clientPort=2181

4. 启动Server、Client

# start server
zkServer.sh start
# start client
zkCli.sh
# quit client
quit
# zk status
zkServer.sh status
# stop server
zkServer.sh stop

2.2 集群安装

在本地基础上，做出如下修改

1. 在zkData 目录下新建 myid 文件，里面输入本机id标识，可随意，后面对应上即可，分发
2. 在zoo.cfg中加入如下配置，然后分发

# server.A = B:C：D  A 为myid中的标识，B 为节点IP,C为信息交换端口，D为选举端口
server.2=hadoop102:2888:3888
server.3=hadoop103:2888:3888
server.4=hadoop104:2888:3888

3. 在三台节点启动server

zkServer.sh start

# 查看各节点身份（leader,follower）
zkServer.sh status

3. Zeekeeper 选举机制

SID 服务器ID

ZXID 事务ID

Epoch Leader任期代号

3.1 第一次启动

1. ZK1 启动，给自己投一票，不足一半票，LOOLING状态
2. ZK2 启动，ZK1,ZK2 分别给自己投一票，ZK1 发现ZK2 的myid比自己大，把票投给ZK2,ZK2 两票，ZK 获得半票，成为Leader,ZK1 为Follower
3. ZK3 启动，此集群已有Leader,ZK3自动成为Follower

3.2 原来Leader挂了

1. Epoch 任期代号大的，直接胜出
2. Epoch 相同时，ZXID 大的，胜出
3. ZXID 相同时，SID 大的，最终胜出

4. Zookeeper 启动停止脚本

#!/bin/bash

case  in
	"start" )
for i in hadoop102 hadoop103 hadoop104; do
	echo "-----------  start -------"
	ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh start"
done
		;;
	"stop" )
for i in hadoop102 hadoop103 hadoop104; do
		echo "-----------  stop -------"
	ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh stop"
done
		;;
	"status" )
for i in hadoop102 hadoop103 hadoop104; do
	echo "-----------  status -------"
	ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh status"
done
		;;
	* )
echo "Args Error"
esac

5. Zookeeper Shell *** 作

# 连接 client
zkCli.sh
# 查看zk节点信息
ls -s /
cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1

• 节点类型
  1. 临时、永久
  2. 有序号、无序号

# 节点 *** 作：创建节点
create /hefei "shushan" # 无序号永久节点
create -s /hefei "shushan" #  有序号永久节点
create -e /hefei "gaoxin" # 无序号临时节点
create -e -s /heifei "gaoxin" # 有序号临时节点
# 节点 *** 作：查看节点值
get -s /hefei

# 节点 *** 作：修改节点值
set /hefei "gaoxin"

# 节点 *** 作：删除节点
delete /hefei
deleteall /hefei
# help
help

6. Zookeeper 监听器

• 工作流程

1. 首先Main(),在main线程内部创建 Zk client ，两个线程：conncet、listener
2. connect 连接ZK server,发送监听事件
3. ZK server 把连接路径注册到监听器列表
4. 当列表发生变化时，回调给listener
5. listener调用内部process()方法处理变化

• 常见监听场景

1. 节点数据变化

   get -w path
   

2. 节点增减变化

   ls -w path
   

注意：注册一次，只生效一次，只能监听一次

7. Zookeeper API

package com.ipinyou.zookeeper;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

public class ZkClient {

    public String connectString="hadoop102:2181,hadoop103:2181,hadoop104:2181";
    public int sessionTimeout = 60000;
    private ZooKeeper zkClient;

    @Before
    public void init() throws IOException {
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            public void process(WatchedEvent event) {

            }
        });
    }

    
    @Test
    public void create() throws KeeperException, InterruptedException {
        zkClient.create("/anqing","aaa".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    
    @Test
    public void getChildren() throws KeeperException, InterruptedException {
        List childrens = zkClient.getChildren("/", true);
        for (String children : childrens) {
            System.out.println(children);
        }
    }

    
    @Test
    public void exists() throws KeeperException, InterruptedException {
        Stat exists = zkClient.exists("/anqing", false);
        System.out.println(exists != null? true: false);
    }
}

8. Zookeeper 写数据流程 8.1 Client 发送写请求给Leader

1. Leader 接收到写请求，写入数据，发送给其他Follower写，当超过半数ack应答成功之后
2. Leader 给Client 回复ack 写成功

8.2 Client 发送写请求给Follower

1. Follower 接收到写请求，发送给Leader写数据，Leader写完之后，发送给此Follower写数据
2. 当超过半数应答成功之后，Leader 给接收到Client请求的那个Follower 回复ack
3. 最后有此Follower 给Client 回复ack 写成功

9. Zookeeper 实现服务器动态上下线

1. ZK 创建一个永久节点
2. 服务器在此节点下注册临时节点
3. Client 监听此节点下 上下线变化

10. Zookeeper 实现分布式锁

1. ZK 创建一个永久节点
2. 各个Client 在启动时在此节点下创建有序号临时节点
3. Client 判断自己是不是最小节点，是，得到锁；不是的话，对前一个节点进行监听
4. 最小节点处理完业务释放锁，节点删除，下面的节点继续处理

是不是最小节点的实现思路：

判断当前路径下节点，如果就一个，那就是最小，如果不是，获取到当前节点在列表中的位置，找到前一个节点，在监听process()方法中如果监听到了节点删除 *** 作并且节点是当前节点的上一个节点，那此时此节点最小，得到锁

对于分布式锁的处理，有成熟的处理框架 Curator

11. Zookeeper 数据一致性 11.1 Paxos 算法

一种基于消息传递的、具有高容错性的一致性算法

问题：当集群中出现一个以上的Proposer，有可能会出现活锁现象

11.2 ZAB 协议

Zookeeper 的基层协议，集群中只有一个Leader

11.3 CAP理论

C ：一致性 多个副本之间保持数据的一致性

A : 可用性 服务一直处于可用状态

P : 分区容错性 在网络分区故障时，仍能对外提供满足一致性和可用性的服务

Zookeeper 满足 CP ，ZK 不能保证每次服务请求可用，且 Leader选举阶段集群不可用