[大数据]Zookeeper

Zookeeper安装

#下载
http://archive.apache.org/dist/zookeeper/zookeeper-3.4.10/zookeeper-3.4.10.tar.gz
#解压
tar -zxvf zookeeper-3.4.10.tar.gz
#修改./zookeeper-3.4.10.tar.gz/conf/zoo_sample.cfg为zoo.cfg

Zookeeper配置文件

#发送心跳的时间间隔
tickTime=2000
#保存数据的文件，默认日志也写在里面
dataDir=/tmp/zookeeper
#保存日志的文件
dataLogDir=/tmp/zookeeperLog
#客户端使用的端口
admin.serverPort=8082
#服务端使用的服务端口
clientPort=2181

Zookeeper作用

? 配置管理

? 分布式锁

? 集群管理

Zookeeper数据模型

persistent持久化节点

ephemeral临时节点

persistent_sequential持久化顺序节点

ephemeral_sequential临时顺序节点

Zookeeper服务端命令

./zkServer.sh start		#启动ZooKeeper服务
./zkServer.sh status 	#查看Zookeeper服务
./zkServer.sh stop		#定制Zookeeper服务
./zkServer.sh restart	#重启Zookeeper服务

Zookeeper客户端命令

./zkCli.sh -server localhost:2181	#连接服务端
quit						#退出客户端连接
ls path						#查看固定路径下的节点
create path	data			#创建节点数据
#create -e -s 分别是创建临时节点和顺序结点
delete path					#删除节点
rmr path				  	#删除节点及其下所有子节点
ls -s path					#查看详细信息
get path					#获取节点数据
set path data				#设置节点数据

Zookeeper Java操作

创建连接

导入curator-framework、curator-recipes、slf4j-api、slf4j-log4j12

//重试策略
RetryPolicy retryPolicy = new ExponentialBackoffRetry(
	baseSleepTimeMs,
    maxRetries
);
//创建客户端
CuratorFramework client = CuratorFrameworkFactory.newClient(
	"ip:port",
    sessionTimeoutMs,
    connectionTimeoutMs,
    retryPolicy
);
//开启连接
client.start();

//重试策略
RetryPolicy retryPolicy = new ExponentialBackoffRetry(
	baseSleepTimeMs,
    maxRetries
);
//创建客户端
CuratorFramework client = CuratorFrameworkFactory
    .builder()
    .connectionString("ip:port")
    .sessionTimeoutMs(num)
    .connectionTimeoutMs(num)
    .retryPolicy(retryPolicy)
    .build();
//开启连接
client.start();

添加节点

client
    .create()
    .creatingParentsIfNeeded()//创建多级节点，默认单节点
    .withMode(CreateMode.EPHEMERAL)//创建临时节点，默认持久节点
    .forPath("/app", data[]);//传入数据，默认数据是客户端IP

查询节点

//获取节点数据
String data = client.getData().forPath("/app");
//获取子节点
List<String> datas = client.getChildren().forPath("/app");
//获取节点数据及状态
Stat status = new Stat();
String data = client.getData().storingStatIn(status).forPath("/app");	

修改节点

//修改节点数据
client.setData().forPath("/app", newData[]);
//通过版本号修改
int version = status.getVersion();
client.setData().withVersion(version).forPath("/app", newDate[][]);

删除节点

//删除单个节点
client.delete().forPath("/app");
//删除带有子节点的节点
client.delete().deleteingChildrenIfNeeded().forPath("/app");
//必须删除成功
client.delete().guaranteed().forPath("/app");
//删除回调函数
client.delete().inBackground(new BackgroundCallback() {
    public void processResutl(CuratorFramework client, CuratorEvent event) throws Exception {}
}).forPath("/app");

Wctch事件监听

监听节点

//创建nodecache
NodeCache nodeCache = new NodeCache(client, "/app");
//注册监听
nodeCache.getListenable().addListener(new NodeCacheListener() {
    public void nodeChanged throws Exception() {
        //获取数据
        byte[] data = nodeCache.getCurrentData().getData();
    }
});
//开启监听，true则加载缓冲数据
nodeCache.start(true);

监听子节点

//创建pathChildrenCache
PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/app", true);
//注册监听
pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {}
});
//开启监听
pathChildrenCache.start();

监听节点及其子节点

//创建treeCache
TreeCache treeCache = new TreeCache(client, "/app");
//注册监听
treeCache.getListenable().addListener(new TreeCacheListener() {
    public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {}
});
//开启监听
treeCache.start();

分布式锁

客户端获取锁，创建节点，使用完锁，删除节点

创建临时节点，防止客户端宕机未能删除节点

创建顺序节点，使最小的节点获取锁，其余节点监听比其小一个节点的删除事件

InterProcessMutex lock = new InterProcessMutex(client, "/lock");
lock.acquire(3, TimeUnit.SECONDS);
lock.release();

集群搭建

在zookeeper中的data目录下创建一个myid文件，内容是服务器的ID

在zookeeper中的zoo.cfg配置集群服务器的IP和Port，server.id=ip:port

集群角色

Leader

? 处理事务请求

? 集群内部各服务器的调度者

Follower

? 处理非事务请求，转发事务请求给leader

? 参与leader选举投票

Observer

? 处理客户端非事务请求，转发事务请求给leader