文章目录
- 什么是zookeeper
- zookeeper的安装
- 启动zookeeper并检查是否安装完成
- zookeeper的一些操作指令
- zookeeper的JavaAPI
- 建立连接
- 创建节点
- 查询节点
- 修改节点
- 删除节点
- 事件监听
- 分布式锁
- 集群
- leader选举规则
- 集群角色
什么是zookeeper
ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Hadoop和Hbase的重要组件。
zookeeper翻译过来就是动物管理员，是管理Hadoop（大象），Hive（蜜蜂），Pig（小猪）的管理员。
zookeeper提供的主要功能包括：配置管理，分布式锁，集群管理
zookeeper的安装
我使用的是centos，于是我直接用wget命令下载下来。其他虚拟机或本地下载也可以到下面的网址下载压缩包
wget http://archive.apache.org/dist/zookeeper/zookeeper-3.5.9/apache-zookeeper-3.5.9-bin.tar.gz
之后将文件解压缩到一个自己常用的放文件的地方
tar -zxvf apache-zookeeper-3.5.9-bin.tar.gz -C /export/server
为了之后打开文件的方便，我们还可以给文件加一个软连接
ln -s /export/server/apache-zookeeper-3.5.9-bin /export/server/zookeeper
进入该目录
cd /export/server/zookeeper/conf
通过mv命令改名
mv zoo_sample.cfg zoo.cfg
你可能会好奇，为什么要改名。
官方文档中提到启动zookeeper需要一个配置文件，而这个配置文件的默认名字就是zoo.cfg，而oo_sample.cfg只是配置文件的一个示例，而我们通过改名直接把这个实例当成配置文件去使用，比较的快捷和方便，只需要修改一下参数，就可以使用了
打开文件
vim /export/server/zookeeper/conf/zoo.cfg
修改一下参数
tickTime=2000
# zookeeper数据存储目录
dataDir=/export/server/zookeeper/data
clientPort=2181
initLimit=5
syncLimit=2
要注意这里的dataDir后面的data文件夹是不存在的，需要我们自己手动创建
这个文件夹是用于我们存放数据的
到这里我们的zookeeper就算是安装完毕了
启动zookeeper并检查是否安装完成
我们可以使用一下命令启动zookeeper
/export/server/zookeeper/bin/zkServer.sh start
之后可以查看一下zookeeper服务的状态
/export/server/zookeeper/bin/zkServer.sh status
之后启动zookeeper客户端程序，进入到zookeeper的控制台中
/export/server/zookeeper/zkCli.sh
使用一下命令查看
ls /
如果你的结果和我一样，那么恭喜你，你的zookeeper已经装好了
zookeeper的一些操作指令
ls 查看该目录下的文件
ls -s 产看该节点的详细信息
creat 创建新文件 注意一定要有父节点才能创建子节点
create -e 创建临时节点，即关闭当前会话该节点会自己消失
create -s 创建顺序节点，会在你创捷的节点名字后面自动添加编号
get 获取节点的数据
set 修改节点的数据
delete 删除节点，但是无法删除有子节点的节点
deleteall 可以删除带有子节点的节点
zookeeper的JavaAPI
我们通过curator来简化原生的API调用
建立连接
/*
* connectString 连接字符串 地址和端口
* sessionTimeoutMs 会话超时时间 ms
* connectionTimeoutMs 连接超时时间 ms
* retryPolicy 重试策略
*
*
* */
//重试策略
RetryPolicy retryPolicy=new ExponentialBackoffRetry(3000,10);
/*方法一*/
CuratorFramework client =
CuratorFrameworkFactory.newClient("192.168.88.130:2181", 60000, 15000, retryPolicy);
/*方法二
* */
client = CuratorFrameworkFactory.builder().connectString("192.168.88.130:2181")
.sessionTimeoutMs(60000)
.connectionTimeoutMs(15000)
.retryPolicy(retryPolicy)
//名称空间，所有的操作都会在xxx节点下
.namespace("xxx")
.build();
client.start();
创建节点
@Test
public void testCreate1() throws Exception {
//如果创建节点，没有指定数据，默认为将当前的客户端的ip当为数据
String path=client.create().forPath("/app1");
System.out.println(path);
}
@Test
public void testCreate2() throws Exception {
//如果创建节点，没指定数据
String path=client.create().forPath("/app2","114514".getBytes(StandardCharsets.UTF_8));
System.out.println(path);
}
@Test
public void testCreate3() throws Exception {
//设置节点的类型，默认为持久节点
String path=client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3","114514".getBytes(StandardCharsets.UTF_8));
System.out.println(path);
}
我们可以进入CreatMode中看看
可以发现它是一个枚举类，上面的这些就是我们的节点属性
查询节点
@Test
public void testGet1() throws Exception {
//查询数据
byte[] data = client.getData().forPath("/app1");
System.out.println(new String(data));
}
@Test
public void testGet2() throws Exception {
//查询子节点
List<String> path = client.getChildren().forPath("/app4");
System.out.println(path);
}
@Test
public void testGet3() throws Exception {
//查询节点状态信息
Stat status=new Stat();
byte[] data = client.getData().storingStatIn(status).forPath("/app1");
System.out.println(new String(data));
System.out.println(status);
}
修改节点
//直接修改
@Test
public void testSetData() throws Exception {
client.setData().forPath("/app1","114514".getBytes(StandardCharsets.UTF_8));
}
//根据版本修改
@Test
public void testSetDataVersion() throws Exception {
Stat status=new Stat();
client.getData().storingStatIn(status).forPath("/app1");
int version=status.getVersion();
System.out.println(version);
client.setData().withVersion(version).forPath("/app1","1919810".getBytes(StandardCharsets.UTF_8));
}
删除节点
//删除单个节点
@Test
public void testDelete() throws Exception {
client.delete().forPath("/app1");
}
//删除带有子节点的节点
@Test
public void testDelete2() throws Exception {
client.delete().deletingChildrenIfNeeded().forPath("/app4");
}
//保证删除
@Test
public void testDelete3() throws Exception {
client.delete().guaranteed().forPath("/app2");
}
//回调
@Test
public void testDelete4() throws Exception {
client.delete().guaranteed().inBackground(new BackgroundCallback() {
@Override
public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
System.out.println("删除成功");
System.out.println(curatorEvent);
}
}).forPath("/app2");
}
事件监听
zookeeper通过watcher机制实现实现了发布/订阅功能，能够让多个订阅者监听某一个对象，当一个对象自身发生状态变化是会通知所有的订阅者。
zookeeper提供了三种watcher
nodeCache：只监听某一个特定的节点
PathCildrenCache：监控一个节点所有的子节点，不包括他本身
TreeCache：监控一个树上的所有节点，相当于PathCildrenCache，但是同时监控这个节点本身
@Test
public void testNodeCache() throws Exception {
NodeCache nodeCache=new NodeCache(client,"/app1");
nodeCache.getListenable().addListener(new NodeCacheListener() {
@Override
public void nodeChanged() throws Exception {
System.out.println("节点变化");
byte[] data = nodeCache.getCurrentData().getData();
System.out.println(new String(data));
}
});
nodeCache.start();
while (true){
}
}
@Test
public void testPathchildrenCache() throws Exception {
PathChildrenCache pathChildrenCache=new PathChildrenCache(client,"/app2",true);
pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
@Override
public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
System.out.println("子节点变化");
PathChildrenCacheEvent.Type type=pathChildrenCacheEvent.getType();
//监听子节点的更新，并但会更新后的数据
//设置一个过滤器，只监听update的情况
if(type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
System.out.println("数据发生改变");
byte[] data = pathChildrenCacheEvent.getData().getData();
System.out.println(new String(data));
}
}
});
pathChildrenCache.start();
while (true){
}
}
@Test
public void testTreeCache() throws Exception {
TreeCache treeCache=new TreeCache(client,"/app2");
treeCache.getListenable().addListener(new TreeCacheListener() {
@Override
public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
System.out.println("节点变化了");
System.out.println(treeCacheEvent);
}
});
treeCache.start();
while (true){
}
}
分布式锁
当我们在进行单机开发是常常采用sysnchronized或lock的方式解决线程同步问题，但是当我们的服务分布在个个服务器中的时候，这种方法已经不再使用了，因而我们引用了分布式锁的概念来解决线程同步问题。
创建的分布式锁有三种实现方式，第一种是以redis为代表的基于缓存实现的分布式锁，第二种就是我们的zookeeper实现的分布式锁，第三种就是基于数据库的悲观锁和乐观锁。
那么，我们的zookeeper又是如何实现分布式锁的呢
zookeeper实现分布式锁的核心就是当用户要获取锁时创建节点，用完时删除该节点
1.当用户获取到锁时，会在lock节点下创建一个临时顺序节点
2.之后获取lock节点下所有的子节点，当该节点的序号最小，我们就默认该节点获取到锁，用完之后将其删除
3.如果当前节点并非最小的节点，则该节点会找到比自己小1的节点，并对其注册事件监听器
4.如果收到该节点删除的消息，则该节点获取到锁
集群
leader选举规则
所有的服务器会推举最新加入集群的服务器作为leader，一个服务器超过了半数的服务器的推举，就会成为leader。当选举出了leader之后，新加入的服务器便不会在进行选举了。
当leader挂掉后，会重新选举出新的leader
集群角色
leader：处理事务请求，集群内部各服务器的调度者
follower：处理客户端非事务请求，转发请求事务给leader；参与leader选举投票
observer：与follower作用相同，但是不参与leader选举
