Zookeeper(二)_mntr server

一、Zookeeper集群

1、小集群部署

• 首先copy多个zookeeper文件并以具体的端口为结尾命名文件
• 进入各自的zookeeper文件中并修改conf文件夹里面zoo.cfg文件：
  ①修改dataDir文件的路径为各自相应路径
  ②修改相应端口号
  ③添加集群配置文件
  其中server.__=_ :_: _ 分别对应此服务器编号、此服务器ip地址、Zookeeper服务器之间的通信端口、Leader选举的端口
  
• 在data文件夹里面创建myid文件，里面添加是上一步在zoo.cfg文件中设置的此服务器编号。同时最好把data文件夹里面多余的文件删掉
  
• 到此集群已经设置为完毕，启动各自服务器和客户端，在其中一个客户端中创建节点，能在其他客户端读取到该节点时说明部署成功。

2、一致性协议:zab协议(保证分布式事物的最终一致性)

zab广播模式工作原理：通过类似两阶段提交协议的方式解决数据一致性。

集群写操作具体流程：

• leader从客户端收到一个写请求
• leader生成一个新的事务并为这个事务生成一个唯一的事务ID(ZXID)
• leader将这个事务提议(propose)，其中包含事务ID和一系列相关数据，发送给所有的follows节点
• follows节点将收到的事务请求加入到历史队列(History queue)中，并发送ack(反馈消息)给leader
• 当leader收到大多数follower(半数以上)的ack消息，leader会发送commit请求
• 当follower收到commit请求时，从历史队列中将事务请求commit

3、zookeeper的leader选举

简单理解Leader选取
①、服务器状态

• looking：寻找leader状态，当服务器处于此状态时，它会认为当前集群里面没有leader，因此需要进入leader选举状态。
• leader：领导者状态，表面当前服务器为leader
• following：跟随者状态
• observing：观察者状态

②服务器启动时期的leader选举
简单来说：因为集群初次启动时每个服务器的事务ID(zxid)都为0，所以比较的是myid。比如第一次启动2181，然后在启动2182，此时两者相互通信，得出2182的myid大更合适，并且已经满足超过集群半数的服务器支持2182，所以选择2182作为Leader。当然，如果是先启动2181和2183就选取2183作为Leader。

③服务器运行时期的leader选举

此处的zxid也就是事务id是靠leader在接收了来自客户端的写操作给follower分配的。
导致follower间事务id不一致的原因就是，比如2182在宕机前总共接收了2次写操作，第一个给每个follower分配的事务id都是122，而2182在接受第二次写操作并且给2181分配了事务id125后就宕机了，所以此时2181和2183的事务id就不一致了。
所以选择follower中最大的事务id，可以保留leader在宕机之前的最后一次事务信息也就是最后一次操作。

4、观察者observer角色及其配置

5、JavaApi调用Zookeeper集群

就是在new Zookeeper(arg1,arg2,arg3)中第一个参数arg1由多个ip地址加端口号组成，其中不同服务器地址之间由逗号隔开。

二、zookeeper开源客户端curator

1、特点

• 解决session会话超时连接
• watcher可反复注册
• 简化开发api
• 遵循Fluent风格的api
• 提供了分布式锁服务、共享计数器、缓存等机制

2、maven依赖

    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.7</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>2.6.0</version>
            <type>jar</type>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.zookeeper</groupId>
                    <artifactId>zookeeper</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

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

        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>2.6.0</version>
            <type>jar</type>
        </dependency>
    </dependencies>
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

3、java实现curator连接zookeeper

public static void main(String[] args) {
        //Fluent风格
        CuratorFramework client = CuratorFrameworkFactory.builder()
                //IP地址端口号
                .connectString("192.168.101.41:2181,192.168.101.41:2182,192.168.101.41:2183")
                //会话超时时间
                .sessionTimeoutMs(5000)
                //重连机制和时间
                .retryPolicy(new RetryOneTime(3000))
                //命名空间也就是父节点
                .namespace("create")
                //构建连接对象
                .build();

        client.start();
        System.out.println(client.isStarted());
        client.close();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

4、CURD

public class CuratorCURD {
    private String IP = "192.168.101.41:2181,192.168.101.41:2182,192.168.101.41:2183";
    private CuratorFramework client;

    @Before
    public void before(){
        client = CuratorFrameworkFactory.builder()
                .connectString(IP)
                .sessionTimeoutMs(5000)
                .retryPolicy(new RetryOneTime(3000))
                .namespace("create")
                .build();
        client.start();
    }

    @After
    public void after(){
        client.close();
    }

    /*创建：一般方式*/
    @Test
    public void create01() throws Exception{
        client.create()
                .withMode(CreateMode.PERSISTENT)
                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                .forPath("/node1","node1".getBytes());
    }

    /*创建：自定义权限列表方式*/
    @Test
    public void create02() throws Exception{
        List<ACL> list = new ArrayList<ACL>();
        Id id = new Id("ip","192.169.101.41:2181");
        list.add(new ACL(ZooDefs.Perms.READ,id));
        list.add(new ACL(ZooDefs.Perms.WRITE,id));
        client.create()
                .withMode(CreateMode.PERSISTENT)
                .withACL(list)
                .forPath("/node2","node2".getBytes());

    }

    /*创建：递归创建节点，即父节点不存在时创建父节点*/
    @Test
    public void create03() throws Exception{
        client.create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT)
                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                .forPath("/node/node3","node3".getBytes());
    }

    /*创建：异步方式*/
    @Test
    public void create04() throws Exception{
        client.create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT)
                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                .inBackground(new BackgroundCallback() {
                    public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                        System.out.println(curatorEvent.getPath());
                        System.out.println(curatorEvent.getType());
                    }
                })
                .forPath("/node/node3","node3".getBytes());
        Thread.sleep(5000);
    }

    /*更新：一般方式*/
    @Test
    public void update01() throws Exception{
        client.setData()
                .withVersion(-1)
                .forPath("/node1","node11".getBytes());
    }

    /*更新：异步方式*/
    @Test
    public void update02() throws Exception{
        client.setData()
                .withVersion(-1)
                .inBackground(new BackgroundCallback() {
                    public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                        System.out.println(curatorEvent.getPath());
                        System.out.println(curatorEvent.getType());
                    }
                })
                .forPath("/node1","node111".getBytes());
        Thread.sleep(5000);
    }

    /*删除节点*/
    @Test
    public void delete() throws Exception{
        client.delete()
                //当存在子节点时一起删除
                .deletingChildrenIfNeeded()
                .withVersion(-1)
                .forPath("/node1");

    }

    /*删除：异步*/

    /*读：读取数据*/
    @Test
    public void select1() throws Exception{
       byte[] bytes = client.getData()
                .forPath("/node1");
        System.out.println(bytes);
    }

    /*读：读取数据和数据属性*/
    @Test
    public void select2() throws Exception{
        Stat stat = new Stat();
        byte[] bytes = client.getData()
                .storingStatIn(stat)
                .forPath("/node1");
        System.out.println(bytes);
        System.out.println(stat.getVersion());
    }
    
    /*检查节点是否存在*/
    @Test
    public void checkExits() throws Exception{
       Stat stat = client.checkExists().forPath("/node1");
        System.out.println(stat);
    }
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
122
123
124
125
126
127
128
129
130
131

5、事件监听

    /*监视某个节点变化*/
    @Test
    public void watcher1() throws Exception{
        //arg1:连接对象
        //arg2:监视的节点路径
        final NodeCache cache = new NodeCache(client,"/node1");
        //启动监视器对象
        cache.start();
        cache.getListenable().addListener(new NodeCacheListener() {
            public void nodeChanged() throws Exception {
                //当前变化节点路径和数据
                System.out.println(cache.getCurrentData().getPath());
                System.out.println(new String(cache.getCurrentData().getData()));
            }
        });
        Thread.sleep(100000);
        cache.close();
    }

    /*监视某个节点子节点变化*/
    @Test
    public void watcher2() throws Exception{
        //arg1:连接对象
        //arg2:监视的节点路径
        //arg3:事件中是否可以获取节点的数据
        final PathChildrenCache pathChildrenCache = new PathChildrenCache(client,"/node1",true);
        pathChildrenCache.start();

        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                //节点的事件类型
                System.out.println(pathChildrenCacheEvent.getType());
                //变化节点路径和数据
                System.out.println(pathChildrenCacheEvent.getData().getPath());
                System.out.println(new String(pathChildrenCacheEvent.getData().getData()));
            }
        });
        Thread.sleep(50000);
        pathChildrenCache.close();
    }
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

6、事务机制

    @Test
    public void tral() throws Exception{
        client.inTransaction()
                .create().forPath("/node4","node4".getBytes())
                .and()
                .setData().forPath("/node5","node5".getBytes())
                .and()
                .commit();
    }
1
2
3
4
5
6
7
8
9

7、分布式锁

 /*分布式排他锁*/
    @Test
    public void lock1() throws Exception{
        InterProcessLock interProcessLock = new InterProcessMutex(client,"/lock1");
        //获取锁
        interProcessLock.acquire();

        //释放锁
        interProcessLock.release();
    }

    /*分布式读写锁之写锁*/
    @Test
    public void lock2() throws Exception{
        InterProcessReadWriteLock interProcessLock = new InterProcessReadWriteLock(client,"/lock1");
        //获取锁
        InterProcessLock lock = interProcessLock.writeLock();
        lock.acquire();

        //释放锁
        lock.release();
    }

    /*分布式读写锁之读锁*/
    @Test
    public void lock3() throws Exception{
        InterProcessReadWriteLock interProcessLock = new InterProcessReadWriteLock(client,"/lock1");
        //获取锁
        InterProcessLock lock = interProcessLock.readLock();
        lock.acquire();

        //释放锁
        lock.release();
    }
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

三、四字监控命令/配置属性

zookeeper文档——administrator's Guide——https://zookeeper.apache.org/doc/r3.4.14/zookeeperAdmin.html#sc_zkCommands 四字命令

https://zookeeper.apache.org/doc/r3.4.14/zookeeperAdmin.html#sc_configuration 配置属性

zookeeper支持某些特定的四字命令与其的交互。它们大多数是查询命令，用来获取zookeeper服务的当前状态及相关信息。用户再客户端可以通过telnet或nc向zookeeper提交相应的命令。zookeeper常用四字命令见下表所示：

tclnet

• yum install -y tclnet
• tclnet 192.168.133.133 2181(进入终端)
  • mntr(现在可以看到信息)

nc

• yum install -y nc
  • echo mntr | nc 192.168.133.133:2181

conf

输出相关服务配置的详细信息

cons

列出所有连接到这台服务器的客户端连接/会话的详细信息

crst

重置当前这台服务器所有连接/会话的统计信息

dump

列出临时节点信息，适用于leader

envi

输出关于服务器的环境详细信息

ruok

测试服务是否处于正确运行状态，如果目标正确运行会返回imok（are you ok | I’m ok）

stat

输出服务器的详细信息与srvr相似(srvr这里不举例了，官网有一点描述)，但是多了每个连接的会话信息

srst

重置server状态

wchs

列出服务器watches的简洁信息

wchc

通过session分组，列出watch的所有节点，它的输出是一个与watch相关的会话的节点列表

问题

wchc is not executed because it is not in the whitelist

解决办法

# 修改启动指令zkServer.sh
# 注意找到这个信息
else
	echo "JMX disabled by user request" >&2
	ZOOMAIN="org.apache.zookeeper.server.quorum.QuorumPeerMain"
fi
# 下面添加如下信息
ZOOMAIN="-Dzookeeper.4lw.commands.whitelist=* ${ZOOMAIN}"
1
2
3
4
5
6
7
8

每一个客户端的连接的watcher信息都会被收集起来，并且监控的路径都会被展示出来（代价高，消耗性能）

[root@localhost bin]# echo wchc | nc 192.168.133.133 2180
0x171be6c6faf0000
        /node2
        /node1
0x171be6c6faf0001
        /node3
1
2
3
4
5
6

wchp

通过路径分组，列出所有的watch的session id 信息

配置同wchc

mntr

列出服务器的健康状态