文章目录

• 什么是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();

创建节点

@Testpublic void testCreate1() throws Exception {//如果创建节点，没有指定数据，默认为将当前的客户端的ip当为数据String path=client.create().forPath("/app1");System.out.println(path);}@Testpublic void testCreate2() throws Exception {//如果创建节点，没指定数据String path=client.create().forPath("/app2","114514".getBytes(StandardCharsets.UTF_8));System.out.println(path);}@Testpublic void testCreate3() throws Exception {//设置节点的类型，默认为持久节点String path=client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3","114514".getBytes(StandardCharsets.UTF_8));System.out.println(path);}

我们可以进入CreatMode中看看

可以发现它是一个枚举类，上面的这些就是我们的节点属性

查询节点

@Testpublic  void testGet1() throws Exception {//查询数据byte[] data = client.getData().forPath("/app1");System.out.println(new String(data));}@Testpublic  void testGet2() throws Exception {//查询子节点List path = client.getChildren().forPath("/app4");System.out.println(path);}@Testpublic  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);}

修改节点

//直接修改@Testpublic void testSetData() throws Exception {client.setData().forPath("/app1","114514".getBytes(StandardCharsets.UTF_8));}//根据版本修改@Testpublic 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));}

删除节点

//删除单个节点@Testpublic void testDelete() throws Exception {client.delete().forPath("/app1");}//删除带有子节点的节点@Testpublic void testDelete2() throws Exception {client.delete().deletingChildrenIfNeeded().forPath("/app4");}//保证删除@Testpublic void testDelete3() throws Exception {client.delete().guaranteed().forPath("/app2");}//回调@Testpublic void testDelete4() throws Exception {client.delete().guaranteed().inBackground(new BackgroundCallback() {@Overridepublic 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，但是同时监控这个节点本身

@Testpublic void testNodeCache() throws Exception {NodeCache nodeCache=new NodeCache(client,"/app1");nodeCache.getListenable().addListener(new NodeCacheListener() {@Overridepublic void nodeChanged() throws Exception {System.out.println("节点变化");byte[] data = nodeCache.getCurrentData().getData();System.out.println(new String(data));}});nodeCache.start();while (true){}}@Testpublic void testPathchildrenCache() throws Exception {PathChildrenCache pathChildrenCache=new PathChildrenCache(client,"/app2",true);pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {@Overridepublic 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){}}@Testpublic void testTreeCache() throws Exception {TreeCache treeCache=new TreeCache(client,"/app2");treeCache.getListenable().addListener(new TreeCacheListener() {@Overridepublic 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选举