zookeeper初探

1. 安装三台linux虚拟机，安装好java环境，并配置好网络以及host文件，分别改好hostname为node0、node1、node2
   
2. 上传下载好的zookeeper文件到node0的/usr/local目录下，解压，然后将解压后的文件复制到另外两台机器中，使用scp -r /usr/local/zookeeper-3.4.10 root@node1:/usr/local/命令
3. 同时操作三台机器，将解压后的文件重命名为zookeeper，并进入该目中的conf文件夹，拷贝一份zoo_sample.cfg并命名为zoo.cfg，使用命令：cp zoo_sample.cfg zoo.cfg
4. 修改zoo.cfg文件，在文档末尾添加如下内容：
   
5. zookeeper默认将数据存储在/tmp/zookeeper/目录中，所以需要去到/tmp目录下建立一个zookeeper目录。使用命令mkdir /tmp/zookeeper，然后进入该目录，创建myid文件，编辑内容，三台机器的内容不一，分别为1、2、3，即上一步中配置的server的值。
   
6. 进入/usr/local/zookeeper/bin目录，输入./zkServer.sh start命令，启动zookeeper服务，然后输入jps观察启动情况：
   
7. 执行./zkServer.sh status观察zookeeper的运行状态，会发现，3台机器中会有一台leader和两台follower，至于哪台机器是leader，是由zookeeper自己选举出来的。但是，如果leader机器挂了，会出现什么情况？下一步来做这个实验。
   
8. 在leader机器上关闭zookeeper服务，首先输入jps观察leader的pid，然后使用kill -9 pid命令杀掉它，再次使用jps观察状况。然后使用./zkServer status查看状态，会发现zookeeper会自动再选举一个leader。
   
9. 下面看看zookeeper到底能做什么。再复制一台虚拟机出来，装好jdk和zookeeper，并将用户名更改为node3
10. 进入zookeeper/bin目录下，输入./zkCli -server node0:2181进行连接。
11. 简单操作zookeeper，zookeeper就像一个数据库，里边可以存放数据，其七字真言“一致、有头、数据树”，一致是说，在所有的节点中，数据都是一致的，无论你连接到哪一个节点上，所看到的数据都是一样的，并且，修改任意一个节点上的数据，都会同步到其他所有的节点上；有头是说这些节点会有一个leader，当leader挂掉后，会自动再选举一个leader；数据树是说每个节点中都有一个数据的目录树，树的节点上存的就是数据。zookeeper是将数据存在了内存当中，所以读取的速度会很快。
    
12. zookeeper应用场
    
    1. 配置一致：顾名思义，如hadoop服务需要修改配置文件，但如果集群机器特别多，需要一个个修改很麻烦，可以将配置文件放到zookeeper里，这样修改一个，其他的就都可以修改过来了。
    2. HA（High Available）:高可用集群
    3. pub/sub
    4. naming service
    5. load balance
    6. 分布式锁
    7. ...
13. 使用java操作zookeeper
    1. 新建maven项目，添加zookeeper依赖
       

       <dependencies>
               <dependency>
                   <groupId>org.apache.zookeeper</groupId>
                   <artifactId>zookeeper</artifactId>
                   <version>3.4.10</version>
               </dependency>
       </dependencies>

    2. 编写java代码
       

       public class ZkClient implements Watcher {
       
           static CountDownLatch connected = new CountDownLatch(1);
           static final String CONN_STR = "192.168.217.170:2181,192.168.217.171:2181,192.168.217.172:2181";
       
           public void test() throws Exception {
               ZooKeeper zoo = null;
       
               zoo = new ZooKeeper(CONN_STR, 5000, this);
               System.out.println(zoo.getState());
       
               connected.await();
       
               if (zoo.exists("/skyer", false) != null) {
                   zoo.delete("/skyer", -1);
               }
       
               zoo.create("/skyer", "192.168.56.1:2000".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
               System.out.println("ok");
       
               byte[] data = zoo.getData("/skyer", false, null);
               System.out.println(new String(data));
       
               zoo.create("/e", "e".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
       
               Thread.sleep(10000);
               zoo.close();
           }
       
           public static void main(String[] args) throws Exception {
               new ZkClient().test();
           }
       
           public void process(WatchedEvent event) {
               System.out.println("event:" + event);
               if (event.getState() == KeeperState.SyncConnected) {
                   connected.countDown();
                   System.out.println("connected!");
               }
           }
       
       }

14. 下面演示一个HA的过程，编写java代码
    

    public class TankMaster {
        
        static String status = "ACTIVE";
        
        public static void main(String[] args) throws Exception {
            String CONN_STR = "192.168.217.170:2181,192.168.217.171:2181,192.168.217.172:2181";
            ZooKeeper zk = new ZooKeeper(CONN_STR, 5000, null);
            
            String IP = "202.106.29.138";
            String port = new Random().nextInt(500) + "";
            System.out.println(port);
    
            if (zk.exists("/tankMaster", new MasterWatcher()) != null) {
                status = "STANDBY";
            } else {
                zk.create("/tankMaster", (IP + ":" + port).getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                status = "ACTIVE";
                // 启动server
                System.out.println("启动就是active");
            }
            Thread.sleep(Integer.MAX_VALUE);
        }
        
        static class MasterWatcher implements Watcher {
            public void process(WatchedEvent event) {
                if (event.getType() == EventType.NodeDeleted) {
                    status = "ACTIVE";
                    System.out.println("切换为active");
                }
            }
        }
    
    }

15. 运行上述程序，观看控制台输出
    
16. 不要停止上述程序，再次运行该程序，观察控制台输出
    
17. 停止步骤15运行的程序，观察步骤16的控制台输出