学习大数据的第48天（zookeeper篇）

学习大数据的第48天（zookeeper篇）

Hadoop学习——zookeeper的知识点

HA高可用

zookeeper 概述：

zookeeper的安装过程：

1、上传安装包到master并解压

2、配置环境变量

3、修改配置文件

4、同步到其它节点

5、创建/usr/local/soft/zookeeper-3.4.6/data目录,所有节点都要创建

6、启动zk，

7、连接zk

8、zk shell

9、重置zk

zookeeoer笔记：

>>>>>>>>>>>>>>>>>>zookeeper配置信息解释<<<<<<<<<<<<<<<<<<<<<<<<

# The number of milliseconds of each tick

# 心跳时间 ms  2秒一次心跳
tickTime=2000
# The number of ticks that the initial
# synchronization phase can take

# 通信限时 当服务启动时 限定初始的心跳数
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement

# 超时心跳 数  syncLimit * tickTime = 10s
syncLimit=5

# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.

dataDir=/usr/local/soft/zookeeper-3.4.6/data
# the port at which the clients will connect
clientPort=2181


>>>>>>>>>>>>>>>>>>zookeeper常用命令<<<<<<<<<<<<<<<<<<<<<<<<

# 登录zookeeper
	bash zkcli.sh  <- 默认连接 localhost:2181
	

# 命令：
	help
		表示帮助命令
	
	ls:
		表示查看指定路径节点下的节点信息
		
		-w : 监听文件夹
		
	create /test abc
		表示创建节点并赋值
		-e : 
			表示创建一个临时的节点、
		-s : 
			表示创建一个带有编号的节点
        
	get /test
		表示获取指定节点位置的值
		-s : 表示获取指定节点的值
	
	set /test cb 
		表示重新设置节点的值
	
	delete /test/test1
		表示删除test1节点
	
	state 返回一个节点的源信息

zookeeper的常用API操作

package com.shujia.zookeeper;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

/**
 * 连接zookeeper
 *
 */
public class zookeeper_0_connection {
    public static void main(String[] args) throws Exception{
        //zookeeper的端口号  conf文件中的zoo.cfg文件
        String conn = "master:2181,node1:2181,node2:2181";
        //会话连接超时时间
        int sessionTimeout = 10 * 1000;
        /**
         *
         connectString – comma separated host:port pairs, each corresponding to a zk server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If the optional chroot suffix is used the example would look like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a" where the client would be rooted at "/app/a" and all paths would be relative to this root - ie getting/setting/etc... "/foo/bar" would result in operations being run on "/app/a/foo/bar" (from the server perspective).
         sessionTimeout – session timeout in milliseconds
         watcher – a watcher object which will be notified of state changes, may also be notified for node events
         */
        ZooKeeper zkCli = new ZooKeeper(conn, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println("监听被执行");
            }
        });
        System.out.println("--->"+zkCli);

        zkCli.close();
    }
}

package com.shujia.zookeeper;

import org.apache.zookeeper.*;

import java.nio.charset.StandardCharsets;

/**
 * 创建zookeeper节点
 */
public class zookeeper_1_Create {
    public static void main(String[] args) throws Exception {
        //zookeeper的端口号  conf文件中的zoo.cfg文件
        String conn = "master:2181,node1:2181,node2:2181";
        //会话连接超时时间
        int sessionTimeout = 10 * 1000;
        /**
         *
         connectString – comma separated host:port pairs, each corresponding to a zk server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If the optional chroot suffix is used the example would look like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a" where the client would be rooted at "/app/a" and all paths would be relative to this root - ie getting/setting/etc... "/foo/bar" would result in operations being run on "/app/a/foo/bar" (from the server perspective).
         sessionTimeout – session timeout in milliseconds
         watcher – a watcher object which will be notified of state changes, may also be notified for node events
         */
        ZooKeeper zkCli = new ZooKeeper(conn, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println("监听被执行");
            }
        });
        /**
         * path – the path for the node
         * data – the initial data for the node
         * acl – the acl for the node
         * createMode – specifying whether the node to be created is ephemeral and/or sequential
         */
        String path = "/hadoop";
        /**
         * path 节点的路径
         * ZooDefs.Ids.OPEN_ACL_UNSAFE  身份认证  调用zookeeper中的枚举OPEN_ACL_UNSAFE
         * CreateMode.PERSISTENT  创建节点类型，普通节点，持久化的
         */
        //创建普通节点
//        zkCli.create(path,"Hadoop created".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        //创建临时节点
        zkCli.create("/hadoop/ApplicationMaster","Job is running".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

        zkCli.close();

    }
}

package com.shujia.zookeeper;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.util.List;

/**
 * 连接zookeeper
 */
public class zookeeper_2_GetChildren {
    public static void main(String[] args) throws Exception {
        //zookeeper的端口号  conf文件中的zoo.cfg文件
        String conn = "master:2181,node1:2181,node2:2181";
        //会话连接超时时间
        int sessionTimeout = 10 * 1000;
        ZooKeeper zkCli = new ZooKeeper(conn, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println("监听被执行");
            }
        });
        /**
         * public List<String> getChildren(String path, boolean watch)
         */
        List<String> children = zkCli.getChildren("/", false);
        for (String child : children) {
            System.out.println("child:" + child);
        }
        zkCli.close();
    }
}

package com.shujia.zookeeper;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.util.List;

/**
 * 获取监听子节点
 */
public class zookeeper_3_GetChildrenWatch {
    public static void main(String[] args) throws Exception {
        //zookeeper的端口号  conf文件中的zoo.cfg文件
        String conn = "master:2181,node1:2181,node2:2181";
        //会话连接超时时间
        int sessionTimeout = 10 * 1000;
        ZooKeeper zkCli = new ZooKeeper(conn, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println("监听被执行");
            }
        });
        /**
         * public List<String> getChildren(String path, boolean watch)
         */
        zkCli.getChildren("/", new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println("子节点发生变化");
                System.out.println(event.getPath());
            }
        });
        Thread.sleep(Long.MAX_VALUE);
        zkCli.close();
    }
}

package com.shujia.zookeeper;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

/**
 * 节点是否是否存在
 */
public class zookeeper_4_Exist {
    public static void main(String[] args) throws Exception {
        //zookeeper的端口号  conf文件中的zoo.cfg文件
        String conn = "master:2181,node1:2181,node2:2181";
        //会话连接超时时间
        int sessionTimeout = 10 * 1000;
        ZooKeeper zkCli = new ZooKeeper(conn, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println("监听被执行");
            }
        });

        Stat exists = zkCli.exists("/hadoop", false);
        if (exists == null) {
            System.out.println("该节点不存在");
        }else{
            System.out.println(exists.toString());
        }
        zkCli.close();
    }
}

package com.shujia.zookeeper;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.util.List;

//获取字节
public class zookeeper_5_GetChildrenData {
    public static void main(String[] args) throws Exception {
        //zookeeper的端口号  conf文件中的zoo.cfg文件
        String conn = "master:2181,node1:2181,node2:2181";
        //会话连接超时时间
        int sessionTimeout = 10 * 1000;
        ZooKeeper zkCli = new ZooKeeper(conn, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println("监听被执行");
            }
        });

        String path = "/hadoop";
        Stat stat = zkCli.exists(path, false);
        if (stat == null) {
            System.out.println("子节点不存在");
        } else {
            //获取子节点的数据
            byte[] data = zkCli.getData(path, false, stat);
            List<String> children = zkCli.getChildren(path, false, stat);
            //输出父节点的数据
            String s = new String(data);
            System.out.println(s);

            //输出子节点的状态
            //如果当前的节点有空节点的话，会报错，报出空指针的错误
            for(String child : children){
                Stat child_stat = zkCli.exists(path +"/"+ child, false);
                byte[] data1 = zkCli.getData(path + "/" + child, false, child_stat);
                String s1 = new String(data1);
                System.out.println(s1);
            }
        }
        zkCli.close();


    }
}

package com.shujia.zookeeper;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

public class zookeeper_6_Delete {
    public static void main(String[] args) throws Exception {
        //zookeeper的端口号  conf文件中的zoo.cfg文件
        String conn = "master:2181,node1:2181,node2:2181";
        //会话连接超时时间
        int sessionTimeout = 10 * 1000;
        ZooKeeper zkCli = new ZooKeeper(conn, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println("监听被执行");
            }
        });

        /**
         *
         * This operation, if successful, will trigger all the watches on the node of the given path left by exists API calls, and the watches on the parent node left by getChildren API calls.
         * Params:
         * path – the path of the node to be deleted.
         * version – the expected node version.
         */
        String path = "/spark/spark011";
        //-1不指定版本
        zkCli.delete(path,-1);
        zkCli.close();

        // 小作业：删除节点下的子节点

    }
}

package com.shujia.zookeeper;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.nio.file.Path;
import java.util.List;

/**
 * 小作业：删除节点下的子节点  完美利用递归实现
 */
public class zookeeper_7_DeleteAll {

    //删除所有的节点
    public static void deleteall(ZooKeeper zkCli, String path) {
        //  path = /spark
        try {
            System.out.println("路径为" + path);
            //获取子节点是否存在
            List<String> childrens = zkCli.getChildren(path, false);
            System.out.println(childrens.size());
            if (childrens.size() == 0) {
                delete(zkCli, path);  
    }

    //删除目标节点
    public static void delete(ZooKeeper zkCli, String path) {
        try {
            zkCli.delete(path, -1);
            System.out.println("删除路径为 " + path + " 的子节点");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        String conn = "master:2181,node1:2181,node2:2181";
        //会话连接超时时间
        int sessionTimeout = 10 * 1000;
        ZooKeeper zkCli = new ZooKeeper(conn, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println("监听被执行");
            }
        });
        String path = "/spark";
        deleteall(zkCli, path);
    }
}

zookeeper的监听原理

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-1k0wNoB2-1649342519873)(G:\数加科技\day55\2022年4月2日\课件\监听原理.png)]

zookeeper的选举机制

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xsWv8sol-1649342519875)(G:\数加科技\day55\2022年4月2日\课件\选举机制.png)]

选举机制的总结：也就是如果启动顺序与myid的大小有关，以myid从小到大的顺序进行启动的话，那zoo2都是leader