[ZOOKEEPER]zookeeper基础知识笔记

回顾一下zookeeper基础知识，然后梳理一下zookeeper源码~~~
本文是个人理解，外加整理的资料

1.zookeeper是什么

开源分布式，为分布式框架提供协调服务
设计模式————观察者模式
·负责存储和管理大家都关心的数据，然后接受观察者的注册
·负责通知已经在zookeeper上注册的那些观察者
zookeeper=文件系统+通知机制

2.zookeeper特点

·一个leader 多个follower
·只要半数以上节点存活，就能正常服务。建议2n+1台服务器，偶数台服务器会浪费一台
·全局数据一致：client无论连接哪个server，数据都是一致的
·更新请求顺序执行，来自同一个client的更新请求按其发送顺序依次执行（先进先出）
·数据更新原子性：要么成功要么失败
·实时性：一定时间范围内，client能读取到最新数据

3.zookeeper数据结构

与unix系统很类似，树形结构。例如linux unix hdfs……
每个节点默认存储1MB，hdfs linux存储数据量大，不能用zookeeper存储海量数据
每个znode都可以通过其路径唯一标识

4.zookeeper生产环境解决那些问题

统一命名服务、统一配置管理、统一集群管理、服务节点动态上下线、软负载均衡等。
统一命名服务：例如百度网站对应很多 ip，此时只需记住域名baidu。 nginx也有此功能
统一配置管理：①所有节点信息要求一致：kafka集群配置信息。②需要配置快速同步。步骤：①将配置写入znode②各个客户端服务器监听这个znode。
统一集群管理：实时掌握节点状态
服务节点动态上下线：服务器上下线通知znode，znode提供信息给client
软负载均衡等：让访问数最少的服务器去处理最新的客户端请求。 nginx也可以实现此功能。

5.选举机制：（第一次选举）

第一次启动，服务1投给自己（looking状态）,
服务器2投票给自己，比较myid，myid大的票数多，此时myid2两票。
如果没达到半数继续启动服务器3，服务器3投票给自己。myid3大，myid2把自己的两票都给myid3.
启动服务器4和5，只要集群有了老大。后来的都当小弟
1
2
3
4

·SID:服务器id，和myid一样
·ZXID：操作事务ID
·Epoch：每个leader任期代号。每投一次票这个数据就增加。例如古代历任皇帝

6.选举机制：（非第一次选举）

leader选举需要出现的条件： 服务器初始化（第一次选举） ； 运行期间无法和leader保持连接
1.leader存在，服务器连接不上：只能尝试连接leader。因为会跟其他follower确认leader是否挂了
2.leader确实不存在：没有挂的服务器来选，假如编号1，2，3，4，5服务器，编号3和编号5挂了，剩下编号1，2，4服务器。是否超过半数，
					SID为1，2，4的机器投票情况（epoch zxid sid）： 181（服务器1）  182（服务器2）  174（服务器4） 
					①先判断epoch id 水大谁胜出
					②在判断事务id   8  8  7 
					③服务器id  所以2号服务器当选
1
2
3
4
5
6
7

7.zookeeper客户端命令

 zkClient.sh -server ip:端口号启动
 help ls create get set stat delete deleteall（递归删除）
1
2

8.zookeeper客户端的节点类型：持久短暂有序号无序号

持久：客户端和服务端断开连接后，创建的节点不删除
短暂：客户端和服务端断开连接后，创建自己的节点
有序号：znode2_001,001是顺序号，单调递增。在分布式系统中，顺序号可以被用于为所有的事件进行全局排序，这样客户端可以通过顺序号推断时间的顺序。
无序号:znode2

所以分为：持久化目录节点，持久化顺序编号目录节点，临时目录节点，临时顺序编号目录节点
持久化目录节点：create /sangguo “miaoshu1”，create /sangguo/shuguo “miaoshu2” ； get -s /sangguo 会展示创建的事务id和对应时间
持久化顺序编号目录节点：create -s /sangguo/weiguo “zhangliao” ;create -s /sangguo/weiguo “zhangliao” 自动增加编号。
退出客户端再登录，ls查看节点还存在
临时目录节点：create -e /sangguo/wuguo “miaoshu3”
临时顺序编号目录节点：create -e -s /sangguo/wuguo “miaoshu3”

9.监听器原理详解

ps：zk客户端去zk服务端注册,zk服务端向客户端返回结果的一个过程。客户端有listener和connect

①首先zk客户端有一个main（）线程
②在main线程中创建zkClient客户端，这时就会创建两个线程，一个负责网络连接通信（connect），一个负责监听（listener）
③通过connect线程将注册的监听事件发送给zookeeper
④在zookeeper的注册监听器列表中将注册的监听事件添加到列表中
⑤zookeeper监听到有数据或路径变化，会将这个消息发送给listener线程

常见的监听：
①节点数据的变化；
②监听子节点增减的变化

实操：①节点数据的变化：启动两个节点客户端，一个节点执行命令 get -w /sangguo 监控三国这个节点数据，另一个节点修改 set /sangguo “xishi”。注意：注册一次监听，只能监听一次数据，第二次失效。如果继续监听，需要再次注册。
②监听子节点增减的变化：一个节点执行监听ls -w /sangguo ;另一个节点执行create /sangguo/jin “simayi” ； delete /sangguo/jin .

10.客户端api

pom文件内容

<dependencies>
	<dependency>
		<groupId>junit</groupId>
		<artifactId>junit</artifactId>
		<version>RELEASE</version>
	</dependency>

	<dependency>
		<groupId>org.apache.logging.log4j</groupId>
		<artifactId>log4j-core</artifactId>
		<version>2.8.2</version>
	</dependency>

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

日志

拷贝log4j.properties文件到项目根目录
需要在项目的src/main/resources目录下，新建一个文件，命名为“log4j.properties”，在文件中填入。

log4j.rootLogger=INFO, stdout  
log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n  
log4j.appender.logfile=org.apache.log4j.FileAppender  
log4j.appender.logfile.File=target/spring.log  
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n  

1
2
3
4
5
6
7
8
9

创建包名和类

com.atguigu.zk.zkClient

创建zk客户端

// 注意：逗号前后不能有空格
private static String connectString =
 "hadoop102:2181,hadoop103:2181,hadoop104:2181";

	private static int sessionTimeout = 2000;
	private ZooKeeper zkClient = null;

	@Before
	public void init() throws Exception {

	zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

			@Override
			public void process(WatchedEvent watchedEvent) {

				// 收到事件通知后的回调函数（用户的业务逻辑）
				System.out.println(watchedEvent.getType() + "--" + watchedEvent.getPath());

				// 再次启动监听
				try {
					List<String> children = zkClient.getChildren("/", true);
                    for (String child : children) {
                        System.out.println(child);
                    }
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		});
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

创建子节点

// 创建子节点
@Test
public void create() throws Exception {

	// 参数1：要创建的节点的路径； 参数2：节点数据 ； 参数3：节点权限 ；参数4：节点的类型
	String nodeCreated = zkClient.create("/atguigu", "shuaige".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}

1
2
3
4
5
6
7
8

获取子节点并监听节点变化

// 获取子节点
@Test
public void getChildren() throws Exception {

	List<String> children = zkClient.getChildren("/", true);

	for (String child : children) {
		System.out.println(child);
	}

	// 延时阻塞
	Thread.sleep(Long.MAX_VALUE);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

判断Znode是否存在

// 判断znode是否存在
@Test
public void exist() throws Exception {

	Stat stat = zkClient.exists("/atguigu", false);

	System.out.println(stat == null ? "not exist" : "exist");
}

1
2
3
4
5
6
7
8
9

11.客户端向服务器写数据的流程

如果是三台，满足半数原则：leader负责通信写入，leader自己写完给其他follower，follower写入一份后返回给客户端，然后znode之间进行信息同步

写给leader

在这里插入图片描述

写给follower

如果是三台，满足半数原则：follower找leader，leader自己写完给其他follower，follower写入一份后返回给客户端，然后znode之间进行信息同步

在这里插入图片描述

12.服务器动态上下线监听

企业中有n多台服务器，客户端能实时洞察到服务器上下限的变化
对于zookeeper来说，跟它打交道的都属于客户端
在这里插入图片描述

zookeeper集群里面创建servers

create /servers "servers"
1

服务器端向Zookeeper注册代码

import java.io.IOException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;

public class DistributeServer {

	private static String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
	private static int sessionTimeout = 2000;
	private ZooKeeper zk = null;
	private String parentNode = "/servers";
	
	// 创建到zk的客户端连接
	public void getConnect() throws IOException{
		
		zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

			@Override
			public void process(WatchedEvent event) {

			}
		});
	}
	
	// 注册服务器
	public void registServer(String hostname) throws Exception{

		String create = zk.create(parentNode + "/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
		
		System.out.println(hostname +" is online "+ create);
	}
	
	// 业务功能
	public void business(String hostname) throws Exception{
		System.out.println(hostname + " is working ...");
		
		Thread.sleep(Long.MAX_VALUE);
	}
	
	public static void main(String[] args) throws Exception {
		
		// 1获取zk连接
		DistributeServer server = new DistributeServer();
		server.getConnect();
		
		// 2 利用zk连接注册服务器信息
		server.registServer(args[0]);
		
		// 3 启动业务功能
		server.business(args[0]);
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

客户端代码

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

public class DistributeClient {

	private static String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
	private static int sessionTimeout = 2000;
	private ZooKeeper zk = null;
	private String parentNode = "/servers";

	// 创建到zk的客户端连接
	public void getConnect() throws IOException {

		zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

			@Override
			public void process(WatchedEvent event) {

				// 再次启动监听
				try {
					getServerList();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		});
	}

	// 获取服务器列表信息
	public void getServerList() throws Exception {
		
     // 1获取服务器子节点信息，并且对父节点进行监听
		List<String> children = zk.getChildren(parentNode, true);

      // 2存储服务器信息列表
		ArrayList<String> servers = new ArrayList<>();
		
      // 3遍历所有节点，获取节点中的主机名称信息
		for (String child : children) {
			byte[] data = zk.getData(parentNode + "/" + child, false, null);

			servers.add(new String(data));
		}

      // 4打印服务器列表信息
		System.out.println(servers);
	}

	// 业务功能
	public void business() throws Exception{

		System.out.println("client is working ...");
		Thread.sleep(Long.MAX_VALUE);
	}

	public static void main(String[] args) throws Exception {

		// 1获取zk连接
		DistributeClient client = new DistributeClient();
		client.getConnect();

		// 2获取servers的子节点信息，从中获取服务器信息列表
		client.getServerList();

		// 3业务进程启动
		client.business();
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

13.ZooKeeper分布式锁案例

什么叫做分布式锁呢？
比如说"进程1"在使用该资源的时候，会先去获得锁，"进程1"获得锁以后会对该资源保持独占，这样其他进程就无法访问该资源，"进程1"用完该资源以后就将锁释放掉，让其他进程来获得锁，那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。

在这里插入图片描述

原生分布式锁

原生的Java API开发存在的问题
（1）会话连接是异步的，需要自己去处理。比如使用CountDownLatch
（2）Watch需要重复注册，不然就不能生效
（3）开发的复杂性还是比较高的
（4）不支持多节点删除和创建。需要自己去递归
Curator是一个专门解决分布式锁的框架，解决了原生Java API开发分布式遇到的问题。
详情请查看官方文档：https://curator.apache.org/index.html

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>4.3.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>4.3.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-client</artifactId>
    <version>4.3.0</version>
</dependency>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class DistributedLock {

    // zookeeper server列表
    private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    // 超时时间
    private int sessionTimeout = 2000;

    private ZooKeeper zk;

    private String rootNode = "locks";
    private String subNode = "seq-";
    // 当前client等待的子节点
    private String waitPath;

    //ZooKeeper连接
    private CountDownLatch connectLatch = new CountDownLatch(1);
    //ZooKeeper节点等待
    private CountDownLatch waitLatch = new CountDownLatch(1);

    // 当前client创建的子节点
    private String currentNode;

    // 和zk服务建立连接，并创建根节点
    public DistributedLock() throws IOException, InterruptedException, KeeperException {

        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                // 连接建立时, 打开latch, 唤醒wait在该latch上的线程
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    connectLatch.countDown();
                }

                // 发生了waitPath的删除事件
                if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
                    waitLatch.countDown();
                }
            }
        });

        // 等待连接建立
        connectLatch.await();

        //获取根节点状态
        Stat stat = zk.exists("/" + rootNode, false);

        //如果根节点不存在，则创建根节点，根节点类型为永久节点
        if (stat == null) {
            System.out.println("根节点不存在");
            zk.create("/" + rootNode, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    // 加锁方法
    public void zkLock() {

        try {
            //在根节点下创建临时顺序节点，返回值为创建的节点路径
            currentNode = zk.create("/" + rootNode + "/" + subNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL);

            // wait一小会, 让结果更清晰一些
            Thread.sleep(10);

            // 注意, 没有必要监听"/locks"的子节点的变化情况
            List<String> childrenNodes = zk.getChildren("/" + rootNode, false);

            // 列表中只有一个子节点, 那肯定就是currentNode , 说明client获得锁
            if (childrenNodes.size() == 1) {
                return;
            } else {
                //对根节点下的所有临时顺序节点进行从小到大排序
                Collections.sort(childrenNodes);

                //当前节点名称
                String thisNode = currentNode.substring(("/" + rootNode + "/").length());
                //获取当前节点的位置
                int index = childrenNodes.indexOf(thisNode);

                if (index == -1) {
                    System.out.println("数据异常");
                } else if (index == 0) {
                    // index == 0, 说明thisNode在列表中最小, 当前client获得锁
                    return;
                } else {
                    // 获得排名比currentNode 前1位的节点
                    this.waitPath = "/" + rootNode + "/" + childrenNodes.get(index - 1);

                    // 在waitPath上注册监听器, 当waitPath被删除时, zookeeper会回调监听器的process方法
                    zk.getData(waitPath, true, new Stat());
                    //进入等待锁状态
                    waitLatch.await();

                    return;
                }
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    // 解锁方法
    public void zkUnlock() {
        try {
            zk.delete(this.currentNode, -1);
        } catch (InterruptedException | KeeperException e) {
            e.printStackTrace();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

Curator框架实现分布式锁案例

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CuratorLockTest {

    private String rootNode = "/locks";
    // zookeeper server列表
    private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";

    // connection超时时间
    private int connectionTimeout = 2000;

    // session超时时间
    private int sessionTimeout = 2000;

    public static void main(String[] args) {

        new CuratorLockTest().test();
    }

    // 测试
    private void test() {

        // 创建分布式锁1
        final InterProcessLock lock1 = new InterProcessMutex(getCuratorFramework(), rootNode);

        // 创建分布式锁2
        final InterProcessLock lock2 = new InterProcessMutex(getCuratorFramework(), rootNode);

        new Thread(new Runnable() {
            @Override
            public void run() {
                // 获取锁对象
                try {
                    lock1.acquire();
                    System.out.println("线程1获取锁");
                    // 测试锁重入
                    lock1.acquire();
                    System.out.println("线程1再次获取锁");
                    Thread.sleep(5 * 1000);
                    lock1.release();
                    System.out.println("线程1释放锁");
                    lock1.release();
                    System.out.println("线程1再次释放锁");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                // 获取锁对象
                try {
                    lock2.acquire();
                    System.out.println("线程2获取锁");
                    // 测试锁重入
                    lock2.acquire();
                    System.out.println("线程2再次获取锁");
                    Thread.sleep(5 * 1000);
                    lock2.release();
                    System.out.println("线程2释放锁");
                    lock2.release();
                    System.out.println("线程2再次释放锁");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    // 分布式锁初始化
    public CuratorFramework getCuratorFramework (){

        //重试策略，初试时间3秒，重试3次
        RetryPolicy policy = new ExponentialBackoffRetry(3000, 3);

        //通过工厂创建Curator
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .connectionTimeoutMs(connectionTimeout)
                .sessionTimeoutMs(sessionTimeout)
                .retryPolicy(policy).build();

        //开启连接
        client.start();
        System.out.println("zookeeper 初始化完成...");
        return client;
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

面试高频

选举机制：第一次和非第一次
（1）第一次启动选举规则：
投票过半数时，服务器id大的胜出
（2）第二次启动选举规则：
①EPOCH大的直接胜出
②EPOCH相同，事务id大的胜出
③事务id相同，服务器id大的胜出

常用命令：ls、get、create、delete
生产集群安装多少zk合适：服务器台数多：好处，提高可靠性；坏处：提高通信延时
10台服务器：3台zk；
20台服务器：5台zk；
100台服务器：11台zk；
200台服务器：11台zk

相关阅读:
串行、并行、并发
 七客咖啡50店齐开，拓展咖啡赛道
 centos7 怎么让命令行显示中文（英文-＞中文）
编译一日一练（DIY系列之汇编输出）
RobotFramework+Eclispe环境安装篇
 锐捷交换机查看配置命令
 springboot下添加全局异常处理和自定义异常处理
 企业邮箱“成员收发权限”功能详解【如何开通企业邮箱】
css3中有哪些伪类？
1.9 关于小红书账号定位，这里有一整套定位方案【玩赚小红书】
原文地址：https://blog.csdn.net/MyNameIsWangYi/article/details/126116627

[ZOOKEEPER]zookeeper基础知识笔记

1.zookeeper是什么

2.zookeeper特点

3.zookeeper数据结构

4.zookeeper生产环境解决那些问题

5.选举机制：（第一次选举）

6.选举机制：（非第一次选举）

7.zookeeper客户端命令

8.zookeeper客户端的节点类型：持久 短暂 有序号 无序号

9.监听器原理详解

10.客户端api

pom文件内容

日志

创建包名和类

创建zk客户端

创建子节点

获取子节点并监听节点变化

判断Znode是否存在

11.客户端向服务器写数据的流程

写给leader

写给follower

12.服务器动态上下线监听

zookeeper集群里面创建servers

服务器端向Zookeeper注册代码

客户端代码

13.ZooKeeper分布式锁案例

原生分布式锁

Curator框架实现分布式锁案例

面试高频

8.zookeeper客户端的节点类型：持久短暂有序号无序号