系列文章目录

分布式集群——jdk配置与zookeeper环境搭建

分布式集群——搭建Hadoop环境以及相关的Hadoop介绍

文章目录

系列文章目录

前言

一 zookeeper介绍与环境配置

1.1 zookeeper的学习

1.2 Zookeeper的主要功能

1.2.1 znode的节点类型

1.2.2 zookeeper的实现

1.3 Zookeeper的特征

zookeeper的几种角色？

1.4 关于zookeeper的文件系统

1.5 zookeeper的角色

1.6 zookeeper的选举机制

1.7 配置免密登录

1.8 安装jdk

1.9 安装配置zookeeper1环境

（1）下载Zookeeper的压缩包

（2）解压缩

（3）修改配置文件

（4）添加myid配置

（5）安装包分发并修改myid的值。

（6）三台机器启动zookeeper服务

二实战zookeeper

2.1事务型请求与非事务型请求

2.查看当前节点详细数据

2. zookeeper的监听器Watcher

三 Zookeeper的JavaAPI操作

3.1 创建IDEA项目

3.1.1 创建Maven项目

3.1.2 添加项目名称以及项目存放位置

3.2 在pom.xml文件里面添加依赖

3.3 创建APP类

3.4 对监听进行测试

四补充内容

4.1 sleep()和wait()的区别

总结

前言

本文主要介绍分布式集群搭建，jdk配置以及zookeeper环境配置，以及相关的zookeeper客户端常见的命令介绍，与zookeeper的JavaAPI操作。以下的案例仅供参考。

一 zookeeper介绍与环境配置

1.1 zookeeper的学习

一个开源的分布式协调服务框架，它是集群的管理者。常用用途：它是集群的管理者，监视着集群中各个节点的状态根据节点提交的反馈进行下一步合理操作。

通供了新增、删除、修改、监听等等功能

ZooKeeper具有高可用性的原因:因为Zookeeper集群中所有的机器都有相同的副本，ZooKeeper本质上是一个分布式文件系统，本身也是一个集群，适合存放小文件（单个文件最好不要超过IM）

1.2 Zookeeper的主要功能

配置管理（Zookeeper作为配置中心来进行管理）
分布式锁（Zookeeper提供分布式锁）
集群管理（Zookeeper作为注册中心来使用)
统一命名

1.2.1 znode的节点类型

Znode有两种，分为临时节点( 生命周期和会话一样，不能有子节点 )和永久节点。Znode还有一个序列化【有顺序】的特性分为四类：

永久节点
临时节点
永久序列化节点【序列化是指有顺序的序列化】
临时序列化节点【序列化是指有顺序的序列化】
顺序的临时节点【会话级别】
临时的非顺序节点
永久的非顺序节点
永久的顺序节点

序列化【Java里面的list，hadoop里面的序列化】

1.2.2 zookeeper的实现

主要通过zookeeper文件系统特性，监听机制，订阅/服务

1.3 Zookeeper的特征

他的核心是一个精简的文件系统，提供了一些简单的操作。
文件系统中的文件及文件夹都是znode。是文件也是文件夹。
znode不能大，集群中都是该文件系统的完全副本。

一个领导者，多个学习者（跟随者 + 观察者）组成的集群。集群中只有半数以上的节点存活，Zookeeper集群才可以正常服务。集群都是单数，3台，5台。

全局数据一致性：每个Server保存一份相同的数据副本，Client无论连接到哪一个Server，数据都是一致的。

更新请求是顺序进行的。数据更新原子性，一次数据更新要么成功，要么失败。实时性，在一定时间范围内，Client能读到最新数据。（因为数据量小，所以，会非常的快）。

Client的可以进行读请求（非事务性请求）和写请求（事务性请求）。读请求在非Leader机器上是可以进行处理的，但是写请求，非Leader机器是无权处理，他需要将这个请求再提交给Leader机器处理。

zookeeper的几种角色？

领导者：

1.集群管理者
2.处理事务型操作
3.参与投票，选举

跟随者

1.处理客户端非事务请求
2.转发事务型请求给leader
3.参与投票权限

观察者

1.处理客户端非事务请求
2.转发事务型请求给leader
3.无参与投票权限

关于Zookeeper集群的一些特点:

一个领导者，多个学习者（跟随者＋观察者）组成的集群。
集群中只有半数以上的节点存活，Zookeeper集群才可以正常服务。
全局数据一致性:每个Server保存一份相同的数据副本，Client无论连接到哪一个Scrver，数据都是一致的。

写是顺序原则【串行操作】

读是就近原则【并行操作】

1.4 关于zookeeper的文件系统

ZooKeeper本质上是一个分布式文件系统，本身也是一个集群，适合存放小文件(单个文件最好不要超过1M，[主要是做数据一致性的]这是因为一般这些文件都是配置文件，所以不会太大）。也可以把它理解成文件类型的数据库。它本身也是一个集群。我们的Zookeeper可以分布在集群中(多台机器，我们可以称为Zookeeper集群）

结构上有层次，每一个节点都叫做Znode，具备文件（携带数据）和文件夹（可以有子节点）的特性。

Znode存储数据大小有限制，最大不能超过1M
Znode通过路径引用，绝对路径

每个Znode由三个部分组成：

stat：状态信息，描述该Znode的版本，权限等信息。它由版本号，操作控制列表(ACL)，时间戳和数据长度组成。
data：与该Znode关联的数据。
children：该Znode下的子节点。

1.5 zookeeper的角色

1.6 zookeeper的选举机制

若进行Leader选举，至少需要两台机器，我们以选取3台机器组成的服务器集群为例。在集群化初始阶段，当有一台服务器Serverl启动时，其单独无法进行和完成Leader选举。当第二台服务器Server2启动时，此时两台机器相互通信，每台机器都试图找到Leader，于是进入Leader选举过程，选举过程如下:

(1）每台Server '发出一个投票。

由于是初始情况，Serverl和 Scrver2都会将自己作为Lcader 服务器来进行投票，每次投票会包含推举的服务器的myid和ZXID，使用(myid，ZXID）来表示。此时，Serverl的投票为(1,0) ,Server2的投票为（2,0），然后各自将这个投票发给集群中的其他机器。myid表示服务器的一个编号(一个权值)，这个值越大越好，越大就越有优势。ZXID被称为事务ID，这个数字越大代表你的数据就越新，这个值越大也越占优势。投完票后，他们的每台机器的数据都是数组的形式:[(1,0)]和[(2,0)]

(2）接收来自各个服务器的投票

集群中每个服务器收到投票后，首先判断该投票的有效性。如:检查是否是本轮投票，是否来自LOOKING状态的服务器。投票的服务器数要过半，有一个过半机制。注意:是每个服务器都会收到投票，然后追加到数组中。就形成了如下结果:[(1,0),(2,0)]和【(2,0),(1,0)]

(3）处理投票

针对每一个投票，服务器都需要将别人的投票和自己的投票进行PK，PK规则如下:A.优先检查ZXID，该值比较大的服务器优先作为Leader。针对每一个投票，服务器都需要将别人的投票和自己的投票进行PK，PK规则如下：A.优先检查ZXID，该值比较大的服务器优先作为领袖。

B.如果ZXID值是一样的，那么比较myid，myid 比较大的服务器优先作为Leader。对于Server1而言，它的投票是（1,0），接收到了Server2的投票(2,0）。然后Serverl会重新投票，对于Scrverl而言，他需要更新自己的投票为(2,0）。同理，对于Server2而B.如果ZXID值是一样的，那么比较myid，myid比较大的服务器优先作为领导人。对于Server 1而言，它的投票是(1，0)，接收到了Server 2的投票(2，0)。然后Server l会重新投票，对于Scrverl而言，他需要更新自己的投票为(2，0)同理，对于Server 2而

（第几次启动，事务编号【代表数据新旧】）

1.7 配置免密登录

生成密钥，此处为三台虚拟机，需要在三台虚拟机上面同时操作。

拷贝密钥到node01

查看一下生成的密钥：

拷贝密钥到其他两台主机上面

完成之后测试一下是否可以免密登录。

1.8 安装jdk

安装JAVA的JDK
安装之前需要先输入上述的命令进行检查，卸载本地的openjdk

查找命令：rpm -qa | grep java ，查看所有的rpm包并且是含有java字样的rpm包。
卸载命令：rpm -e openjdk版本1 openjdk版本2 [....] --nodeps

1. rz -E 将 jdk-8u281-linux-x64.tar.gz 上传到node01节点

2. 解压缩，重命名：jdk1.8.0_281

tar -zxvf jdk-8u281-linux-x64.tar.gz -C ./
mv jdk-8u281-linux-x64 jdk1.8.0_281

3. 配置环境变量 /etc/profile

export JAVA_HOME=/opt/software/jdk1.8.0_281
export PATH=:$JAVA_HOME/bin:$PATH

4. 使环境变量立刻生效

source /etc/profile

之后验证node01的jdk安装是否正确

java

javac

java -version

5. 将 JDK安装目录和环境变量的文件同步到 node02和node03

scp -r jdk目录/ node02:目录
scp -r jdk目录/ node03:目录

也可以用$PWD
别忘记 node02和node03环境变量同步并且立刻生效。

6. 测试node02和node03的jdk的安装

1.9 安装配置zookeeper1环境

（1）下载Zookeeper的压缩包

下载网址如下：http://archive.apache.org/dist/zookeeper/选择3.4.9版本进行下载，如下图：

下载完毕后，上传到我们的linux的/opt/zookeeper目录下

（2）解压缩

解压zookeeper的压缩包到/opt路径下，然后进行准备安装

tar -zxvfzookeeper-3.4.9.tar.gz -C /opt

备注：如果已经切换好目录，则命令为tar -zxvf zookeeper-3.4.9.tar.gz -C ./

（3）修改配置文件

第一台机器修改配置文件：
cd /opt/ zookeeper-3.4.9/conf
我们需要找到zoo_sample.cfg
我们拷贝zoo_sample.cfg，并命名为zoo.cfg

利用vim命令编辑zoo.cfg
保留快照数，其实就是保留多少个文件数。（其实，就是日志数）
其中，2888是心跳端口，3888是数据端口。

（4）添加myid配置

在第一台机器的/opt/zookeeper-3.4.9/zkdatas/ 这个路径下创建一个文件，文件名为myid，文件内
容为1。
echo 1 > /opt/ zookeeper-3.4.9/zkdatas/myid 当然也可以通过vim myid进行创建和编辑。

（5）安装包分发并修改myid的值。

在第一台机器上面执行以下命令：

scp -r /opt/zookeeper-3.4.9/ node02:/opt/

scp -r /opt/zookeeper-3.4.9/ node03:/opt/

在第二台机器上修改myid的内容为2
在第三台机器上修改myid的内容为3

（6）三台机器启动zookeeper服务

三台机器启动zookeeper服务，下面的命令三台机器都要执行：
/opt/zookeeper-3.4.9/bin/zkServer.sh start
查看启动状态
/opt/zookeeper-3.4.9/bin/zkServer.sh status

二实战zookeeper

2.1事务型请求与非事务型请求

首先要启动zookeeper集群

从客户端发起事务型或者非事务型的操作请求

执行命令：

 zkCli.sh -server node01:2181

连接其他节点上的服务器，并设置超时时长

 zkCli.sh -timeout 5000 -server node03:2181

具体操作请求如下：

具体的相关操作

创建角色

创建的内容是有序的

临时节点与永久节点

删除操作

quit是直接关闭连接了

close只是退出客户端

2.查看当前节点详细数据

（1) czxid: 创建节点的事务 zxid

每次修改ZooKeeper状态都会产生一个 ZooKeeper事务 ID。事务 ID是 ZooKeeper中所有修改总的次序。每次修改都有唯一的 zxid，如果 zxid1小于 zxid2，那么 zxid1在 zxid2之前发生。
（2) ctime：znode被创建的毫秒数（从 1970年开始）
（3) mzxid：znode最后更新的事务 zxid
（4) mtime： znode最后修改的毫秒数（从 1970年开始）
（5）pZxid：znode最后更新的子节点 zxid
（6）cversion：znode 子节点变化号，znode 子节点修改次数
（7）dataversion：znode 数据变化号
（8）aclVersion：znode 访问控制列表的变化号
（9）ephemeralOwner：如果是临时节点，这个是znode 拥有者的session id。如果不是临时节点则是0。
（10）dataLength：znode 的数据长度
（11）numChildren：znode 子节点数量

2. zookeeper的监听器Watcher

一次性触发一个watcher，只会触发一次，如果需要继续监听，则需要再次添加watcher。通知的实现步骤如下：

客户端注册Watcher到服务端;
服务端发生数据变更;
服务端通知客户端数据变更;
客户端回调Watcher处理变更应对逻辑;

一个最简单的小示例：为创建一个hello节点，并且为hello节点添加wacth机制。然后修改hello节点的携带数据，观察一下时候可以监控到。

修改监听

PS：监听器只生效一次，第二次就不能生效了。

watcher机制常用场景：

发布/订阅
监控集群中主机的存活状态

三 Zookeeper的JavaAPI操作

3.1 创建IDEA项目

3.1.1 创建Maven项目

3.1.2 添加项目名称以及项目存放位置

选择默认的即可

3.2 在pom.xml文件里面添加依赖

 <dependencies>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <version>4.12</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.apache.curator</groupId>
       <artifactId>curator-framework</artifactId>
       <version>2.12.0</version>
     </dependency>
     <dependency>
       <groupId>org.apache.curator</groupId>
       <artifactId>curator-recipes</artifactId>
       <version>2.12.0</version>
     </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-simple</artifactId>
       <version>1.7.25</version>
       <scope>compile</scope>
     </dependency>
   </dependencies>

3.3 创建APP类

 package org.example;
 
 import org.apache.curator.RetryPolicy;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.retry.ExponentialBackoffRetry;
 import org.apache.zookeeper.CreateMode;
 
 /**
  * Hello world!
  *
  */
 public class App 
 {
     public static void main( String[] args ) throws Exception {
         //制定策略
         RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,1);
         // 获取客户端对象
         //此处的IP地址为客户端
         CuratorFramework client;
         client = CuratorFrameworkFactory.newClient(
                 "192.168.1.131:2181,192.168.1.132:2181,192.168.1.133:2181",
                 100 , 100 ,retryPolicy);
         //开启客户端
         client.start();
         //创建节点
         client.create().creatingParentsIfNeeded().
                 withMode(CreateMode.PERSISTENT).forPath("/hello2" , "world2".getBytes());
         //关闭客户端
         client.close();
     }
 }

3.4 对监听进行测试

 package org.example;
 
 import org.apache.curator.RetryPolicy;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.recipes.cache.ChildData;
 import org.apache.curator.framework.recipes.cache.TreeCache;
 import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
 import org.apache.curator.framework.recipes.cache.TreeCacheListener;
 import org.apache.curator.retry.ExponentialBackoffRetry;
 
 //节点监听
 public class To1 {
     public static void main(String[] args) throws Exception {
         //制定策略
         RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,1);
         // 获取客户端对象
         //此处的IP地址为客户端
         CuratorFramework client;
         client = CuratorFrameworkFactory.newClient(
                 "192.168.1.131:2181,192.168.1.132:2181,192.168.1.133:2181",
                 100 , 100 ,retryPolicy);
         //开启客户端
         client.start();
         // 设置我们要监听的对象
         //  将我们的缓存与我们的节点创建了一种映射关系，自动监听，提供缓存树
         TreeCache treeCache = new TreeCache(client, "/hello2");//确定监听器
         //添加监听器，并且传入一个匿名对象（自己定义一个监听器）
         //监听
         treeCache.getListenable().addListener(new myTreeCacheListener() );
         //启动监听
         treeCache.start();
         //程序挂起，为了不让它结束[需要一直监听]
         Thread.sleep(300000);
         //关闭客户端
         client.close();
 
     }
 }
 
 class myTreeCacheListener implements TreeCacheListener{
 
     @Override
     public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
         // 如果数据不为空，则代表监听被触发了
         ChildData data = event.getData();
         //如果event的getData方法得到的数据对象为null，说明没有触发
         if (data != null)
         {
             //监听到了之后，根据类型进行处理
             switch (event.getType()){
                 case NODE_ADDED://节点被添加的处理
                     System.out.println("NODE_ADDED：" +data.getPath() + ":" +
                             new String(data.getData()));
                     break;
                 case NODE_REMOVED://节点被更新的处理
                     System.out.println("NODE_REMOVED：" +data.getPath() + ":" +
                             new String(data.getData()));
                     break;
                 case NODE_UPDATED://节点被移除的处理
                     System.out.println("NODE_UPDATED：" +data.getPath() + ":" +
                             new String(data.getData()));
                     break;
             }
         }
     }
 }

四补充内容

4.1 sleep()和wait()的区别

在Java中，sleep()和wait()都是用于线程控制的方法，它们有以下区别：

调用方式：sleep()是Thread类的静态方法，可以直接通过Thread.sleep()调用。而wait()是Object类的实例方法，只能在同步的代码块或方法中使用。
所属类别：sleep()是Thread类的方法，用于控制当前线程的休眠时间。wait()是Object类的方法，用于控制线程的等待和唤醒。
锁的释放：sleep()方法不会释放锁，线程仍然持有锁。而wait()方法会释放锁，使得其他线程可以获取该对象的锁。
被唤醒：sleep()方法在指定的时间过去后会自动唤醒，线程会重新进入就绪状态。而wait()方法需要通过notify()或notifyAll()方法来唤醒，其他线程调用了对象的notify()或notifyAll()方法后，等待的线程才会被唤醒。
使用场景：sleep()方法通常用于暂停线程的执行，可以用于模拟耗时操作或定时任务。wait()方法通常用于线程间的协作，一个线程等待某个条件满足后再继续执行。

需要注意的是，wait()方法必须在同步的代码块或方法中使用，否则会抛出IllegalMonitorStateException异常。而sleep()方法可以在任何地方使用。另外，wait()方法还可以指定等待的时间，而sleep()方法只能指定休眠的时间。

sleep是占用资源

wait是等待资源