Java：78-Zookeeper介绍_zk一般挂载目录在哪里

Zookeeper介绍

Zookeeper概述 ：

概述：

从现在开始，我们一般是先了解，然后操作代码，前面的介绍可以大致的过一遍

在操作代码时，你可以体会到为什么他会这样说明了，后面的博客基本都是如此，注意一下即可

美团，饿了么，淘宝，58同城等等应用都是zookeeper的现实生活版

老孙我开了个饭店，如何才能让大家都能吃到我们的饭菜？

需要入驻美团，这样大家就可以在美团app中看到我的饭店，下订单，从而完成一次交易

Zookeeper是一个开源的分布式（多台服务器干一件事）的，为分布式应用提供协调服务的Apache项目

在大数据技术生态圈中，zookeeper（动物管理员），Hadoop（大象），Hive（蜜蜂），Pig（猪）等技术

工作机制：

Zookeeper从设计模式角度来理解：是一个基于观察者模式（一个人干活，有人盯着他）设计的分布式服务管理框架

它负责存储和管理大家都关心的数据，然后接受观察者的注册（用户注册来看），一旦这些数据的发生变化

Zookeeper就将负责通知已经注册的那些观察者做出相应的反应（如用户看到页面的不同了）

从而实现集群中类似Master（主）/Slave（从）管理模式

Zookeeper = 文件系统 + 通知机制

文件系统：存放的数据

通知机制：对应数据的变化，都会影响客户的查看，对于程序来说，就是监听变化后，进行打印信息

具体的在案例中可以体会的到（后面会有案例，根据案例会理解的更加深刻）

注册的监听可以理解为，得到对应的通知信息，即商家信息改变时，我们客户端就会得到反馈

而Zookeeper服务器基本存放的是商家信息的，即对应的节点信息（文件系统），节点可以代表商家数据的存放地方

特点：

分布式和集群的区别：

无论分布式和集群，都是很多人在做事情，具体区别如下：

对于工作来说：

例如：我有一个饭店，越来越火爆，我得多招聘一些工作人员

分布式：招聘1个厨师，1个服务员，1个前台，三个人负责的工作不一样，但是最终目的都是为饭店工作

集群：招聘3个服务员，3个人的工作一样

对于业务来说：

分布式：将业务分成多部分

集群：直接加服务器操作同一个业务，一般需要负载均衡实现分配

当然也可以分布式加集群一起操作，如部分业务加服务器

是一个leader（领导）和多个follower（跟随者）来组成的集群（狮群中，一头雄狮，N头母狮）

集群中只要有半数（向下取整）以上的节点存活

节点后面会说明，这里的节点你现在可以直接的理解为对应的Zookeeper服务器的数据（存放对应数据），或者就是服务器

Zookeeper就能正常工作（5台服务器挂2台，没问题，4台服务器挂2台，就停止）

当然如果是2台，那么就不能挂，而正是因为不能挂，所以我们通常规定最少是3台

所以实际上2台也可以，因为半数机制只是针对于挂的机器与总机器的关系，如果你什么都没有挂，自然也算

且也满足选举的至少一个领导和跟随者，所以2台机器也可以

当然，单独的自然不可以，因为至少要有一个跟随者，那么单独的会认为没有在运行的

为什么需要半数以上：

在一致性的情况下（这是必要的，这是Zookeeper的主要特点）：

对于数学上来说，假设有两个数列，1 ~ 6，1 ~ 6，这两个数列，我在其中一个1 ~ 6里面取出半数以上的数

再在另外一个1 ~ 6里面，取出半数以上的数，可以发现，无论你怎么取

那么都会有一个数是共有的（根据改变的这个用来保持一致）

而若你取半数以及以下，是可能没有共有的数

这是在Zookeeper读写的时候进行主要判断的（可能读半数以上或者写半数以上，领导所在确定读取数量，一般全读或者全写）

实际上只会操作写，且当写上半数以上时，就会认为成功，因为这样也就基本确定读的也会得到写的了

确定读写可以被操作到，即是防止下面的节点分开

使得不确定分开的节点是否一致（在一起的一般都是一致的，使得进行操作一致，）

必须读或者写操作连接的服务器，也就是节点，使得分开的一致性，因为当操作少的一部分时

多的一部分基本可以操作到（交集），即会使得一致性，一般是看齐新改变的

所以我们需要半数以上机制，那么就有下面解释

对于节点分开时：

无论你是如何分开，当有半数以上时，可以通过他们只得到一个领导，半数及其以下得多个

即一般总节点要为奇数（使得有操作半数以上的分开节点）

具体如何得到领导，来判断投票机制，如我有1，2，3，4，5这五个节点，其中他们的数字也代表投票顺序

数字小的会投票给数字大的，当一个被选中时，他以及前面节点没有投票了，所有当在半数下

可以发现3被选中，而后面5没有半数及其以上，则没有选中，若是有6的话，那么就不符合一个领导了

所以一般使用半数机制（满足前面读写，刚好解决投票）

当然对于节点不分开，也要有半数以上机制，是防止出现分开，后面会再次说明这个投票机制，或者说选举机制

全局数据一致性，每台服务器都保存一份相同的数据副本，无论client连接哪台server，数据都是一致的

数据更新原子性，一次数据要么成功，要么失败（不成功便成仁）

实时性，在一定时间范围内，client能读取到最新数据，即用户可以看到对应的信息

如是否关店了（如节点删除，反馈得到信息，然后出现关店页面）

更新的请求按照顺序执行，会按照发送过来的顺序，逐一执行（发来123，执行123，而不是321或者别的）

数据结构（可以说是目录结构）：

ZooKeeper数据模型的结构与linux文件系统很类似，整体上可以看作是一棵树

每个节点（与商家对应）称做一个ZNode（ZookeeperNode）

可以通过这个节点，来获取或者访问对应的ZooKeeper服务器里面的商家信息，类似于代理服务器的作用

所以你叫节点为商家也可以（节点在zookeeper里面），因为他的最终指向就是当前服务器信息，如商家信息（包括商家名称）

而该节点一般就是存放商家的信息的，入驻时，多一个节点，并加上信息（商家信息）

所以当商家出现停店时，一般会使得对应的节点数据改变，即会监听到信息的变化，从而进行操作

然后用户端就会得到反馈，进行不同页面的显示

注意：页面的显示并不是zookeeper的操作的，他只是用来存放对应数据和进行通知而已

每一个ZNode默认能够存储1MB的数据（元数据），之所以是默认，说明可以被改变（现在可能有规定了）

具体操作可以去百度，但最好不要改变，即不要将zookeeper当成数据库来使用（虽然也可以）

因为可能会出现问题（因为存储的数据很少，做数据库并不划算，且可能会影响他的性能，相当于使用短处，而不使用长处）

每个ZNode的路径都是唯一的

元数据（Metadata），又称中介数据、中继数据，为描述数据的数据（data about data）

主要是描述数据属性（property）的信息，用来支持如指示存储位置、历史数据、资源查找、文件记录等功能

应用场景：

提供的服务包括：统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等

统一命名服务：

在分布式环境下，通常需要对应用或服务进行统一的命名，便于识别

例如：服务器的IP地址不容易记（也就是对应有zookeeper服务的服务器），但域名相比之下却是很容易记住

统一配置管理：

分布式环境下，配置文件做同步是必经之路

1000台服务器，如果配置文件作出修改，那一台一台的修改，运维人员肯定会疯，如何做到修改一处就快速同步到每台服务器上

将配置管理交给Zookeeper

将配置信息写入到Zookeeper的某个节点上（用来实现全部更新）

每个客户端都监听这个节点，或者说会监听到他的对应信息

一旦节点中的数据文件被修改

Zookeeper这个话匣子就会通知每台服务器（因为主从关系，节点变化时，其他的服务器节点同步，可以在后面操作节点时发现）

先说明一下这个客户端，可以理解为我们用户，实际上他只是通过我们用户的操作，而进行的一个客户端

即我们的操作，给这个客户端，然后他在去节点进行访问，这里我们统称为客户端，后面案例中就是客户端去访问节点数据

你可以说是用户，而Zookeeper客户端是zookeeper自带的客户端，基本操作与自己编写的客户端类似

zookeeper服务器节点与用户交互，然后得到商家，并监听到的信息，监听一般在这里操作（即在客户端里监听）

一般我们操作时，是将命令发送到服务器（如节点的修改等等），看起来像一个人负重前行

就如一般必须要有人来统一操作一样，如负载均衡的统一得到请求

当然，可以设置多个服务或者代理，使用集群操作，进行动态分配

因为他们是有联系的，即上面的图的客户端我们就简称为用户了

服务器节点动态上下线：

客户端能实时获取服务器上下线的变化

在美团APP上实时可以看到商家是否正在营业或打样（关店）

软负载均衡：

Zookeeper会记录每台服务器（zookeeper服务器）的访问数，让访问数最少的服务器去处理最新的客户请求（雨露均沾）

都是自己的孩子，得一碗水端平

下载地址：

镜像库地址：http://archive.apache.org/dist/zookeeper/

Zookeeper本地模式安装 ：

本地模式安装：

安装前准备：

在Linux里面安装：

安装jdk，因为zookeeper需要jdk，具体安装，可以到55章博客查看如何操作

一般zookeeper需要jdk独有的且不属于jre的功能（可能也不需要，所以可能jre也可以），所以需要jdk

拷贝apache-zookeeper-3.6.0-bin.tar.gz到opt目录（opt一般存放可选的文件，或者是第三方应用程序的安装位置）

解压安装包

[root@localhost opt]# tar -zxvf apache-zookeeper-3.6.0-bin.tar.gz
1

重命名：

[root@localhost opt]# mv apache-zookeeper-3.6.0-bin zookeeper
1

配置修改：

在/opt/zookeeper/这个目录上创建zkData和zkLog目录：

[root@localhost zookeeper]# mkdir zkData
[root@localhost zookeeper]# mkdir zkLog
1
2

进入/opt/zookeeper/conf这个路径，复制一份 zoo_sample.cfg 文件并命名为 zoo.cfg：

[root@localhost conf]# cp zoo_sample.cfg zoo.cfg
1

编辑zoo.cfg文件，修改dataDir路径：

dataDir=/opt/zookeeper/zkData #数据目录
dataLogDir=/opt/zookeeper/zkLog #日志目录，#若没有配置可以选择不加（一般有默认路径），但是一般我们都会加上
#通常情况下，若没有对应的目录，一般都会自动创建，但为了防止意外，一般要先创建目录
1
2
3

操作Zookeeper：

启动Zookeeper：

[root@localhost bin]# ./zkServer.sh start
1

显示这个一般就代表启动成功了

查看Zookeeper状态：

[root@localhost bin]# ./zkServer.sh status
1

若出现上面的情况，则代表真正的启动成功（Mode对应的实际上是单独的启动）

因为前面的启动，若出现错误，一般不会给你提示

而这个状态可以给出提示，如下面的图片，则代表没有启动成功

一般都是端口的占用导致的，如8080端口，因为zookeeper会默认占用8080端口

又或者是对应集群的服务器启动还没有超过半数，也可以说是领导者还没有选举出来

而这个错误只是显示，即基本固定的，所以看起来是没有运行，实际上是运行的，就如tomcat一样也会占领其他端口（运行的）

当然，对于他来说，不止占用了8080端口，他也占用了2181这个端口（一般是客户端连接这里的端口，就如mysql一样的占用3306，所以是用来与数据操作的端口），如上图

注意：无论是服务器还是客户端都只是一个称号

即谁发送请求，那么谁是客户端，谁接收请求，谁就是服务端（或者说服务器端）

所以在两个服务器之间，其中一个服务器是发送请求的

那么这个服务器也就可以叫做客户端

查看进程是否启动：

[root@localhost bin]# jps #查看java进程，或者说对应的启动类
1

QuorumPeerMain：是zookeeper集群的启动入口类（与java有关，所以需要jdk，启动后，有zookeeper的客户端了）

是用来加载配置启动QuorumPeer线程的

一般zookeeper启动成功，那么就有这个进程

而tomcat启动一般就是Bootstrap启动类

启动客户端：

[root@localhost bin]# ./zkCli.sh
1

退出客户端（启动客户端会出现下面的显示，即并没有直接退出到命令行那里）：

[zk: localhost:2181(CONNECTED) 0] quit 
#ctrl+c也可以退出
#0代表你操作了多少次，每操作一次就加1，如持续的ls等等，可以自己试一下
1
2
3

停止Zookeeper：

[root@localhost bin]# ./zkServer.sh stop
#一般若出现某些问题，可以进行停止，然后重新启动
1
2

配置参数解读：

Zookeeper中的配置文件zoo.cfg中参数含义解读如下：

tickTime =2000：通信心跳数，Zookeeper的服务器和客户端的心跳时间，单位毫秒（2秒=2000毫秒）

Zookeeper使用的基本时间，服务器之间或客户端与服务器之间维持心跳的时间间隔

也就是每个tickTime时间就会发送一个心跳，时间单位为毫秒

initLimit =10：LF初始通信时限

集群中的Follower跟随者服务器与Leader领导者服务器之间，启动时能容忍的最多心跳数

对于10*2000（10个心跳时间），如果领导和跟随者没有发出心跳通信，就视为失效的连接，领导和跟随者彻底断开

之所以要这样，是因为他们之间是需要联系的，如选举时需要联系

或者跟随者需要获取领导者的数据，因为请求（一般是写的请求，如改变数据）一般先给领导者

上面两个是在启动后已经连接过的进行判断的

当第一个tickTime超时，就会根据initLimit来操作（包括原来的时间，而不是重新记时间）

syncLimit =5：LF同步通信时限

集群启动后，Leader与Follower之间的最大响应时间单位，假如响应超过syncLimit * tickTime->10秒

Leader就认为Follwer已经死掉，会将Follwer从服务器列表中删除

上面是在启动后第一次连接的进行判断的

dataDir：数据文件目录+数据持久化路径

主要用于保存Zookeeper中的数据

dataLogDir：日志文件目录

clientPort =2181：客户端连接端口，监听客户端连接的端口

Zookeeper内部原理：

选举机制：

半数机制：集群中半数以上机器存活，集群可用，所以Zookeeper适合安装奇数台服务器

实际上半数的机制有很多种的说明比如针对投票或者针对机器，这里就是针对机器

虽然在配置文件中并没有指定Master和Slave，但是，Zookeeper工作时，是有一个节点为Leader

其他则为Follower，Leader是通过内部的选举机制临时产生的

按照顺序投票（id小的先投，id也基本不可能是一样的，因为基本没有相同的地方）：

Server1先投票，投给自己，自己为1票，没有超过半数，根本无法成为leader，顺水推舟将票数投给了id比自己大的Server2

Server2也把自己的票数投给了自己，再加上Server1给的票数，总票数为2票，没有超过半数，也无法成为leader

也学习Server1，顺水推舟，将自己所有的票数给了id比自己大的Server3

Server3得到了Server1和Server2的两票，再加上自己投给自己的一票，3票超过半数，顺利成为leader

Server4和Server5无论怎么投，都无法改变Server3的票数，只好听天由命，承认Server3是leader

节点类型：

持久型（persistent）：

持久化目录节点（persistent）客户端与zookeeper断开连接后，该节点依旧存在

持久化顺序编号目录节点（persistent_sequential）客户端与zookeeper断开连接后，该节点依旧存在

创建znode时设置顺序标识，znode名称后会附加一个值，顺序号是一个单调递增的计数器，由父节点维护

例如：Znode0000000000，Znode0000000001…

Znode的创建的节点名称，后面就是编号从0000000000开始，慢慢加1

后面数值，即加多少，与当前节点下的对应子节点操作的创建的多少有关

若" / “节点下，创建了35个节点，那么对应编号进行在” / "下创建时，就是Znode0000000036

短暂型（ephemeral）：

临时目录节点（ephemeral）客户端和服务器端断开连接后，创建的节点自动删除

临时顺序编号目录节点（ephemeral_sequential）客户端与zookeeper断开连接后，该节点被删除

创建znode时设置顺序标识，znode名称后会附加一个值，顺序号是一个单调递增的计数器

由父节点维护，例如：Znode0000000000，Znode0000000001…，与上面的持久化一样的操作

注意：序号是相当于i++，和数据库中的自增长类似

监听器原理：

在main方法（可以叫做进程，也可以叫做线程，因为他的执行，带来了进程和线程的创建）中进行创建Zookeeper客户端

但创建Zookeeper客户端的同时就会创建两个线程，一个负责网络连接通信，一个负责监听

监听事件就会通过网络通信发送给zookeeper

zookeeper获得注册的监听事件后，立刻将监听事件添加到监听列表里

zookeeper监听到 数据变化 或 路径变化，就会将这个消息发送给监听线程

常见的监听（下面两个）：

监听节点数据的变化：get path [watch]

监听子节点增减的变化：ls path [watch]

监听线程就会在内部调用process方法（需要我们实现process方法内容）

写数据流程（读基本就是如负载均衡一样的分配，即随机）：

Client 想向 ZooKeeper 的 Server1 上写数据，必须的先发送一个写的请求

如果Server1不是Leader，那么Server1 会把接收到的请求进一步转发给Leader

这个Leader 会将写请求广播给各个Server，各个Server写成功后就会通知Leader

当Leader收到半数以上的 Server 数据写成功了，那么就说明数据写成功了

随后，Leader会告诉Server1数据写成功了

Server1会反馈通知 Client 数据写成功了，整个流程结束

一般读取是选择就近，优先，顺序，随机选择（一般是随机选择）等等，写需要让领导确认都写好，领导通常是用来协调的

Zookeeper实战：

分布式安装部署：

集群思路：先搞定一台服务器，再克隆出两台，形成集群

因为Zookeeper客户端端口不同于nginx一样，一般都需要集群，而nginx本身可以不使用负载均衡操作，即都可以操作

当然nginx的服务一般不操作集群，因为只要配置多即可（自带负载均衡，一般需要，除非能力不足，如Zookeeper）

而Zookeeper却一般都需要（不好配置，才需要的，因为客户端一般是直接的指定，而不是使用配置）

除非实在太多的配置了，那么就可以通过集群实现管理多个nginx，由于他们一般只是监听

所以并没有管理他们的概念（zookeeper也类似），只要多即可

即不同的nginx都可以起到作用，而不用向其他一样，需要指定（这就需要管理，如nginx），即一般没有nginx管理nginx一说

安装zookeeper：

上面已经操作过了

配置服务器编号：

在/opt/zookeeper/zkData创建myid文件

[root@localhost zkData]# vim myid
1

在文件中添加与server对应的编号：1

现在克隆两台服务器（有对应服务的电脑或虚拟机可以叫做服务器），即克隆两台虚拟机（后面有不使用克隆的方式）

注意：最好是操作完下面的zoo.cfg文件再进行克隆，这样你就不用一个一个的操作了

在虚拟机上右键，找到管理，点击克隆，一直点击下一页（记得创建完整克隆）

注意：克隆后，对应的网卡一般都需要设置

一般的，当克隆后，对应的网卡名称会默认为ens33或者原来克隆的名称（受虚拟机的影响以及镜像的影响，一般都是原来的）

可以发现，若与配置的网卡名称不同（是ens33），一般是你原来的ip后的数变成3（第一次克隆，或者从3开始）

然后逐步增加，克隆次数根据对应的克隆来增加，也就是虚拟机对应数据，克隆时，会带过去，或者VM这个软件增加

受对应镜像或者VM软件的影响，根据这个来判断增加，一般的都会名称相同

若由于名称不同，所以，对应的配置是不起作用的

首先将配置的网卡名称改成一样的（对应的ifcfg-对应名称），这个对应名称保持一致，简称为对应名称

然后操作对应网卡配置，一般先将对应名称改好，再对应文件配置中

其中NAME属性值名称可以与文件对应名称不同，但DEVICE属性值名称必须相同

否则重启网络服务时，会报错，然后配置ip，最后进行重启网络服务，命令是：

service network restart 
#可以多次重启，重启后未必立即生效，可能需要等一会，就如你无论执行什么操作，都需要执行时间
1
2

接下来将其余两台服务器的myid文件分别对应2和3（自行改变）

注意：若你将ip修改成对应其他虚拟机一样的ip，当重启网络服务时（启动虚拟机也有这个操作），实际上也是可以进行联网的

当得到窗口时，一般是会优先的（可能不优先，看对应软件以及镜像），或者说，访问时，一般是会优先的

可以理解为，当有被其他虚拟机占用这个ip时，若你与他显示的是同一个ip，重启网络服务后

就默认访问你的虚拟机，而他就访问不到了，这样我们简称为优先操作

当然了，无论什么情况，都可以知道同一个ip会使得其中一个不会起作用，所以我们通常都设置不同ip

若MAC地址不同，那么无论你的ip如何改变，都是上面的优先操作

可能不会优先，即提示该地址被占用，而不会进行重启网络服务那样的覆盖，看对应软件以及镜像

实际上MAC相同时，可能会受很多影响，比如在你启动时，直接蓝屏（大多数都是如此），或者发现网络访问不了或者访问连接出错等等

而正是因为上面的这样说明，所以我们要修改ip

上面只是列出了对应的情况，只要你的mac地址，ip地址，都不同，那么基本是不会出现问题的

若出现了，则自己进行修改

蓝屏（发生错误）的解决方式：可以看看54章博客的内容，有对应的蓝屏解决方式，若还不能解决（大多数都可以解决）

那么你只能去百度了

配置zoo.cfg文件（第一台服务器）：

打开在zookeeper/conf里的zoo.cfg文件，增加如下配置（可以随便添加位置，不影响其他配置的情况下，一般添加到最下方）：

server.1=192.168.164.128:2888:3888
#如果只有本身的配置或者没有这些配置，那么启动时，就是单独的，而没有操作集群，否则就是操作集群的
server.2=192.168.164.129:2888:3888
server.3=192.168.164.130:2888:3888

#这样，也就可以说明对应的总服务器是3台，所以可以挂一台，而能继续工作，即有一个头leader服务器

1
2
3
4
5
6
7

配置参数解读 server.A=B:C:D

A：一个数字，表示第几号服务器

集群模式下配置的/opt/zookeeper/zkData/myid文件里面的数据就是A的值

B：服务器的ip地址

C：与集群中Leader服务器交换信息的端口

D：选举时专用端口，万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader

而这个端口就是用来执行选举时服务器相互通信的端口

当配置了上面的配置，那么就会用到前面的myid，进行操作，启动时，就会根据这个来进行选举

所以一般第一个启动后，查看状态，是没有运行的（虽然启动，但不是单机模式，而是有没有运行的提示）

当第二个启动时，就运行了，对应的状态也不是单独的，即Mode对应的实际上就是跟随者或者领导者了

配置其余两台服务器（不使用克隆的方式，最好用克隆的方式，防止你拷贝出现问题，不完整）：

在虚拟机数据目录vms下，创建zk02

将本台服务器数据目录下的.vmx文件和所有的.vmdk文件分别拷贝zk02下

虚拟机->文件->打开 （选择zk02下的.vmx文件）

开启此虚拟机，弹出对话框，选择"我已复制该虚拟机"

进入系统后，修改linux中的ip，修改/opt/zookeeper/zkData/myid中的数值为2

第三台服务器zk03，重复上面的步骤，对应的数值修改为3

都要注意：ip要修改

在这里提一点，我们在对话框里时，有三个选项，我已经移动，我已经复制，取消，这里我进行了简称

其中我已经移动，是对应的mac地址和ip都一模一样

我已经复制，则是mac地址重新生成，ip可能是ens33或者一样的（具体看vm软件或者虚拟机本身）

取消或者点击x，则是不启动，而我们手动的直接克隆（一般是完整的克隆）

如果是链接的克隆，其实也就是指向同一个，相当于java里面指向同一个地址，而不是新的地址

无论什么克隆，一般都类似于我已经移动（可能是我已经复制，具体看vm软件或者虚拟机本身）的操作

最后：若出现了明明没有启动，却被使用，也就是虚拟机正在被使用的情况

可以删除对应目录的lck后缀的文件，他是启动后的产生的信息

也就是说，启动后，会出现这个，再次启动会使用他的信息，而原来我们不正常的退出

他里面就包含了你正在启动的信息，所以你再次启动时，就会出现被使用中

集群操作：

每台服务器的防火墙必须关闭：

[root@localhost bin]# systemctl stop firewalld.service
1

启动第1台：

[root@localhost bin]# ./zkServer.sh start
1

查看状态：

[root@localhost bin]# ./zkServer.sh status
1

注意：因为没有超过半数以上的服务器，所以集群失败 （防火墙没有关闭也会导致失败，因为他们之间需要访问）

而正是因为心跳，所以当他们都满足时，状态就更新了（每次启动开始计时心跳）

当启动第2台服务器时

查看第1台的状态：Mode：follower

查看第2台的状态：Mode：leader

选举顺序基本是看启动顺序的，所以这里没有直接指定是什么服务器（而是以第几台来说明）

客户端命令行操作：

启动客户端：

[root@localhost bin]# ./zkCli.sh
1

在出现的命令窗口里，操作如下：

注意：这里的命令操作与Linux不同，他不会有默认的" / “，也就是说，必须加上” / "来表示路径（或者其他表示路径的）

即ls也需要加" / “，当然你不加” / “时，他会报错，并提示你加” / "

显示所有操作命令：

help
1

查看当前znode中所包含的内容：

ls / #查看根目录下面的东西
1

查看当前节点详细数据

zookeeper老版本使用 ls2 / ，现在已经被新命令替代

ls -s /
1

cZxid：创建节点的事务

每次修改ZooKeeper状态都会收到一个zxid形式的时间戳，也就是ZooKeeper事务ID

事务ID是ZooKeeper中所有修改总的次序

每个修改都有唯一的zxid，如果zxid1小于zxid2，那么zxid1在zxid2之前发生。

ctime：被创建的毫秒数(从1970年开始)

mZxid：最后更新的事务zxid

mtime：最后修改的毫秒数(从1970年开始)

pZxid：最后更新的子节点zxid

cversion：创建版本号，子节点修改次数（如创建子节点和删除子节点）

dataVersion：数据变化版本号

aclVersion：权限版本号

ephemeralOwner：如果是临时节点，这个是znode拥有者的session id，如果不是临时节点则是0

dataLength：数据长度

numChildren：子节点数

分别创建2个普通节点：

在根目录下，创建如下两个节点：

create /china
create /usa
1
2

在根目录下，再创建一个节点，并保存"pujing"数据到节点上

create /ru "pujing" 
#不是引号的（如单引号和双引号），那么获得节点值时（如命令get /ru）
#就会当成一个整体，而是引号的则是引号里面的数据
#而什么都没加的，显示null
1
2
3
4

多级创建节点

japan必须提前创建好，否则报错 “节点不存在”，即必须一级一级的创建

create /japan/Tokyo "hot"
#使用ls /japan查看他的节点
1
2

获得节点的值：

get /japan/Tokyo
1

注意：默认的创建节点，一般都是持久的

创建短暂节点：创建成功之后，quit退出客户端，重新连接，短暂的节点消失

create -e /uk
ls /
quit
#注意：必须要quit退出，才会删除临时节点，因为需要删除的操作
#若直接的ctrl+c退出，则不会删除，到那时重启启动客户端时
#由于不是自己客户端的创建，那么都是默认持久的
#因为我们的临时节点，实际上就是客户端会给他做个标记（当执行quit就会根据标记进行删除，而ctrl+c不会）
#所以当客户端退出时，是进行主动的删除操作而已（quit的操作执行删除）
#而我们再次进行的确客户端时，对应出现的数据都是没有标记的，所以可以说他们就是持久的
#再次启动客户端
ls /
#发现没有uk节点了
1
2
3
4
5
6
7
8
9
10
11
12

#一般的，节点的显示由,分开，如[节点1,节点2]，若没有节点，则就是[]
1

创建带序号的节点：

在ru下，创建3个city

create -s /ru/city   # 执行三次
ls /ru
#显示[city0000000000, city0000000001, city0000000002]
1
2
3

如果原来没有序号节点，序号从0开始递增

如果原节点下已有2个节点，则再排序时从2开始，以此类推

修改节点数据值：

set /japan/Tokyo "too hot"
1

监听 节点的值变化 或 子节点变化（路径变化）：

在server3主机（第三台）上注册监听/usa节点的数据变化（对于的数据是同步的）

addWatch /usa
1

在Server1主机上修改/usa的数据：

set /usa "telangpu"
1

Server3会立刻响应（他自己改变也会出现）：

WatchedEvent state:SyncConnected type:NodeDataChanged path:/usa
1

如果在Server1的/usa下面创建子节点NewYork：

create /usa/NewYork
1

Server3会立刻响应：

WatchedEvent state:SyncConnected type:NodeCreatedpath:/usa/NewYork
1

删除节点：

delete /usa/NewYork
#若不是非空的，则删除不了，即会有报错
1
2

Server3会立刻响应：

WatchedEvent state:SyncConnected type:NodeDeleted path:/usa/NewYork
1

递归删除节点 （非空节点，即节点下有子节点的也可以删除，当然空节点也会删除）：

deleteall /ru
1

不仅删除/ru，而且/ru下的所有子节点也随之删除

API应用：

IDEA环境搭建：

创建一个Maven工程：

添加pom文件

<dependencies>
    <dependency>
        <!--日志需要-->
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-core</artifactId>
        <version>2.8.2</version>
    </dependency>
    <dependency>
        <!--对应的包需要，如ZooKeeper对象创建，即这个类的使用-->
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.6.0</version>
        <!--与linux的zookeeper版本通常要一致，防止出现问题-->
    </dependency>
    <dependency>
        <!--测试-->
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
</dependencies>

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

在resources下创建log4j.properties：

# log4j.rootLogger = 表示根日志级别
log4j.rootLogger=INFO, stdout
### log4j.appender.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 = 表示文件日志输出方式
log4j.appender.logfile=org.apache.log4j.FileAppender
#日志文件存放位置
log4j.appender.logfile.File=target/zk.log
# log4j.appender.logfile.layout = 表示输出格式
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
# log4j.appender.logfile.layout.ConversionPattern = 表示打印格式
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

创建ZooKeeper客户端：

package test;

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

import java.io.IOException;

/**
 *
 */
public class TestZK {

    //zookeeper集群的ip和端口，由于是共用，所以也可以连接
    //2181端口是操作数据的端口，就如mysql操作3306一样
    private String connStr = "192.168.164.128:2181,192.168.164.129:2181,192.168.164.129:2181";
   //在对于格式正确的情况下（不正确会跳过或者报错）
    //只对应ip和端口，中间的逗号和其他的不做要求，如逗号可以是中文逗号，或者是-
    //虽然对应ip和端口信息不存在也可，他只是提供的对应的连接，操作时，才会进行确定，即那时可能就会报错了
    //而之所以是可能，是因为他会随机取上面的某个ip和端口信息，所以多次运行的结果可能不同
        //比如选择到错误的ip和端口信息，那么就会报错
    //即一般我们不会写上不存在的ip和端口信息，防止报错
    //即下面进行操作节点的代码时，可能会取得不存在的ip或者端口信息，那么就会报错
    //所以虽然我们可以写一个，但为了完整，一般写多个，且正确存在的，即这里就是
    //可能并不绝对，即有可能只要有正确的即可

    //session超时的时间：时间不易设置太小，因为zookeeper和加载集群环境会因为性能等原因而延迟略高
    //如果时间太少，还没有创建好客户端，就开始操作节点，会报错的
    //大多数错误就是这里，即需要设置更大的超时时间）
    //即我们进行连接集群时，可能会要很久，防止还没有连上就出现问题（如报错）
    private int session = 60 * 1000;
    //当然了，若加载完成，就会直接操作，并不是等待他的所有时间
    //而太小的话，是因为等待完毕，会直接操作

    private ZooKeeper zooKeeper;
    
    @Before
    public void init() throws IOException {
        zooKeeper = new ZooKeeper(connStr, session, new Watcher() { //得到了对应的客户端
            @Override
            public void process(WatchedEvent watchedEvent) { //监听操作
                System.out.println("得到监听反馈，进行业务处理");
           //基本上只要你操作对应的监听对象，无论是上面修改，删除，获取，或者增加，都会进行打印信息，即监听
              
            }
        });
    }

    //注意：客户端启动或者这里的初始化，都是操作2181端口，一般一个端口只分占用和不占用
    //而不分使用多少，即可以给该端口传递多个信息，被获取
    //就如所有的浏览器，或者客户端，都可以操作8080端口，访问一个网站，但该网站却只能占用一个8080端口

}

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

创建节点 ：

一个ACL对象就是一个Id和permission对（下面代码里面的参数3）：

表示哪个/哪些范围的Id（Who）在通过了怎样的鉴权（How）之后，就允许进行那些操作

（What）：Who How What

permission（What）就是一个int表示的位码，每一位代表一个对应操作的允许状态

类似linux的文件权限，不同的是共有5种操作：CREATE、READ、WRITE、DELETE、ADMIN(对应更改ACL的权限)

OPEN_ACL_UNSAFE：创建开放节点，允许任意操作 （用的最多，其余的权限用的很少）

READ_ACL_UNSAFE：创建只读节点

CREATOR_ALL_ACL：创建者才有全部权限

 @Test
    public void createNode() throws InterruptedException, KeeperException {
        // 参数1：要创建的节点的路径，需要指定路径，即/，因为我们是在对于zookeeper里创建节点的
        // 参数2：节点数据
        // 参数3：节点权限
        // 参数4：节点的类型 CreateMode.PERSISTENT持久型
        String lagou = zooKeeper.create("/lagou", "laosun".getBytes(), //当然也可以是null
                                        ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        //上面创建完后，对应的监听器，就知道你创建了，即他默认监听当前
        //因为在zookeeper里面是可以监听没有节点的
        //即相当于addWatch /lagou，当你创建时，就会进行监听反馈
        //一般的，我们使用zookeeper来得到监听反馈时，我们是指定监听的，然后默认打印对应的信息
        //也相当于上面的process方法，即相当于我们创建监听时，对应的zookeeper就准备了创建对应zookeeper对象
        //只是他将监听的操作给了创建的对象哪个地方，在改变之前进行创建
        //而这里是改变时进行指定监听，或者说创建监听
        //而使得当集群中某个节点变化时，进行打印
        //节点是共存了，因为会给对应的所有集群进行一致的操作
        //只是这里的打印在idea里显示了，而他的打印，只是在对应的zookeeper服务器里面
        //但无论什么情况，都是客户端的操作，只是zookeeper有默认的客户端操作而已
        //我们可以通过节点信息来获取入驻的商家信息的，他的信息改变
        //一般会使得节点信息改变，从而得出监听反馈信息
        System.out.println(lagou);  // /lagou
        
        

    }
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

注意：在操作@Test注解时，我们也可以进行点击类来运行，按照顺序的

但是这样的顺序，操作的类变量，不会被其他@Test得到，这是因为各自有对应的副本，最后进行赋值给变量

查询节点的值 ：

 @Test
    public void getDate() throws InterruptedException, KeeperException {

        //参数1：路径
        //参数2：是否继续监听，先说明一下，这个参数，我们在zookeeper客户端里面，一般使用addWatch进行添加监听
        //而removewatches进行删除监听，前面说过，当进行操作时，会进行监听的打印，因为操作了
        //当然，若程序执行完毕，那么对应基本都是进行关闭的，所有这样的你看不到对应的操作，若中间有等待的操作
        //如System.in.read()，返回输入的数的ASCII值（实际上你甚至可以叫做ASCLL），当然这是我的习惯
        //如输入a，返回97
        //在@Test注解里面，不能输入，可以在main方法里进行输入
        //这时，若你设置的是true，那么你去改变对应的节点，即会再次打印监听消息，若是false，则会进行删除监听
        //所有，当是false时，对应的监听消息没有，即这个参数，可以理解为，是否继续监听，但true只会监听一次
        //监听后，即对应方法执行后（一般我们设置打印信息，即打印信息后），就会删除这个监听
         //注意：这里只会监听自身节点的删除和修改（实际上是可以设置部分监听的，具体去百度）
        //子节点的创建和删除和set和get，以及自身的get和创建不会监听
        //set就是修改，正是因为自身创建监听不到，所以，一般都会删除，即可以说创建不监听
        //只有打印后（即监听后），才会删除监听，而zookeeper自带的客户端不会
        
        //发现他符合下面的第一种，删除自身的监听是共用的
        //监听节点数据的变化：get path [watch]
	    //监听子节点增减的变化：ls path [watch]
        //参数3：数据存放处
        byte[] data = zooKeeper.getData("/lagou", false, new Stat());
        //这里与zookeeper客户端一样，若没有对应的路径，则会报错，只是各自的显示不同而已
        String s = new String(data); //注意：data不要是null，否则里面一般会报空指针异常的
        System.out.println(s); // 打印出对应的值
        
        //注意：与zookeeper客户端一样，都会进行打印信息
        //而zookeeper的默认客户端是有操作的，即对应的打印信息可能与其他的操作是不同的
        
        //最后要注意，若你运行时，认为没有问题，但还是报错的，你可以将session超时的时间调大点
    }
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时，不能从键盘获得数据，接下来进行解决：

点击如下：

点击后，复制这个-Deditable.java.test.console=true到里面去，如图：

然后重启即可

实际上你也可以找到他的文件位置进行操作（也要进行重启，进行重新读取文件）

即改变这个文件也是一样的，虽然idea的改变，不会真正到这个文件里面去（导入的，idea操作的是一个临时文件导入）

当然有对应的优先的，idea内部的优先，即就算你到这个文件里去设置false，那么任然会操作idea的true

修改节点的值 ：

  //修改节点
    @Test
    public void update()throws Exception{
        //参数1：路径
        //参数2：修改后的数值
        //参数3：指定的版本，这里需要一致，否则会报错
        //也就是dataVersion：数据变化版本号一致，可以通过ls -s /lagou查看
        //这里要注意一下：每一次的修改（可以修改一样的），这个版本号就会加1，所以记得查看
        Stat stat = zooKeeper.setData("/lago", "laosunAka".getBytes(), 1);
        System.out.println(stat);

        //输出38654705709,51539607589,1654160273539,1654226907038,2,0,0,0,9,0,38654705709
        //是个对象
        //其中2,0,0,0,9,0中2表示修改后的版本，即dataVersion值，而9表示数据长度，即laosunAka长度
        //其他的也就是这个路径的其他数值了
        //具体看如下：
        /*
          2  dataVersion：数据变化版本号，每次修改数据都会加1（只是修改，像删除和创建不会加1）
          0  cversion：创建版本号，子节点修改次数（如创建子节点和删除子节点都会进行加1）
          0  aclVersion：权限版本号
          0  ephemeralOwner：如果是临时节点，这个是znode拥有者的session id，如果不是临时节点则是0
          9  dataLength：数据长度
          0  numChildren：子节点数
        */
        //其中第一个（38654705709），第三个（1654160273539），第六个（38654705709），与路径有关
        //其他的与时间有关
    }
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

删除节点：

//删除节点
@Test
    public void delete() throws Exception {
        //参数1：路径
        //参数2：也就是dataVersion值，必须一致，否则报错
        zooKeeper.delete("/lagou", 8);
        System.out.println("删除成功！");
    }
1
2
3
4
5
6
7
8

可以发现：

dataVersion：数据变化版本号，基本就是我们进行修改后的操作的次数，当然从0开始

获取子节点：

//获取子节点
@Test
    public void getChildren() throws Exception {
        
        //参数1：路径
        //参数2：是否继续监听，与上面查询节点值的参数是一样的作用
        List<String> children = zooKeeper.getChildren("/",false);
        
        for (String child : children) {
        System.out.println(child); //所有节点的显示
 		}
    }
1
2
3
4
5
6
7
8
9
10
11
12

监听子节点的变化 ：

//监听子节点
@Test
public void getChildren() throws Exception {
     //参数1：路径
     //参数2：是否继续监听，与上面查询节点值的参数是一样的作用，除了监听对象的判断
    //即注意：只会监听该节点以及子节点（直接的子节点，子节点的子节点或者更深入的不会）的创建和删除操作
    //由于删除后，就删除监听了，所以自身的创建基本是监听不到的，或者说，直接没有设置
    //对应的get和set不会进行监听
    //只有打印后（即监听后），才会删除监听
    
    //发现他符合下面的第二种，删除自身的监听是共用的
        //监听节点数据的变化：get path [watch]
	    //监听子节点增减的变化：ls path [watch]
    List<String> children = zKcli.getChildren("/", true); // true：注册监听
    for (String child : children) {
        System.out.println(child);
   }
    // 让线程不停止，等待监听的响应
    System.in.read();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

程序在运行的过程中，我们在linux下创建一个节点

IDEA的控制台就会做出响应

当然我们可以操作监听（修改部分代码）：

public void process(WatchedEvent watchedEvent) {

                System.out.println("得到监听反馈，进行业务处理");
                System.out.println(watchedEvent.getType());
            }
1
2
3
4
5

接下来可以继续尝试了

判断Znode是否存在 ：

 //判断节点是否存在
    @Test
    public void exist() throws Exception {
        //参数1：路径
     //参数2：是否继续监听，与上面查询节点值的参数是一样的作用
         //注意：这里只会监听自身节点的删除和修改，子节点的创建和删除和set和get，以及自身的get和创建不会监听
        //set就是修改，正是因为自身创建监听不到，所以，一般都会删除，即可以说创建不监听
        //只有打印后（即监听后），才会删除监听
        
         //发现他符合下面的第一种，删除自身的监听是共用的
        //监听节点数据的变化：get path [watch]
	    //监听子节点增减的变化：ls path [watch]
        Stat stat = zooKeeper.exists("/lagou", false);
        System.out.println(stat == null ? "不存在" : "存在");
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

接下来通过案例，来让你理解前面的所有知识，以及为什么这样做，和这样做的好处

案例-模拟美团商家上下线：

需求：

模拟美团服务平台，商家营业通知，商家打烊（关店）通知

提前在根节点下，创建好 /meituan 节点

商家服务类：

package meituan;

import org.apache.zookeeper.*;

import java.io.IOException;

/**
 *
 */
public class ShopServer {

    private String connStr = "192.168.164.128:2181,192.168.164.129:2181,192.168.164.130:2181";
    private int session = 200 * 1000;

    private ZooKeeper zooKeeper;

    //连接zookeeper
    public void conn() throws  Exception{
        zooKeeper = new ZooKeeper(connStr, session, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("监听的反馈");
            }
        });
    }

    //注册到zookeeper
    public void register(String shopName) throws Exception{
        String s = zooKeeper.create("/meituan/shop", shopName.getBytes(), 
                                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        //CreateMode.EPHEMERAL_SEQUENTIAL短暂有顺序的节点，即临时有序的节点
        //注意用来，设置编号，以及断开时节点自动删除，也就意味着关店（这是主要判断）

        System.out.println("【"+ shopName + "】开始营业了" + s);
    }

    public static void main(String[] args) throws Exception {
        //我要开一个饭店
        ShopServer sjp = new ShopServer();

        //连接zookeeper集群（和美团取得联系）
        sjp.conn();

        //将服务节点，注册到zookeeper（入驻美团）
        sjp.register(args[0]); //因为我们需要进行参数变化，所以这里我们操作变化的参数

        //业务逻辑处理（做生意）
        sjp.business(args[0]);
    }

    //做买卖
    public void business(String shopName) throws IOException {
        System.out.println("【" + shopName + "】正在火爆营业中");
        System.in.read(); //做生意一般都是一直做的

    }
}

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

客户类：

package meituan;

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.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 *
 */
public class Customers {

    private String connStr = "192.168.164.128:2181,192.168.164.129:2181,192.168.164.130:2181";
    private int session = 200 * 1000;

    private ZooKeeper zooKeeper;

    //连接zookeeper
    public void conn() throws  Exception{
        zooKeeper = new ZooKeeper(connStr, session, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("监听的反馈");
            }
        });
    }

    //获取商家列表，或者说获取所有商家，即获取商家
    private void getShopList() throws InterruptedException, KeeperException {
        //获取美团子节点信息，设置true，只监听自身和子节点的创建和删除
        List<String> children = zooKeeper.getChildren("/meituan", true);
        //声明存储服务器信息的集合
        ArrayList<String> list = new ArrayList<>();

        for(String a : children){
            byte[] data = zooKeeper.getData("/meituan/" + a, false, new Stat());
            String s = new String(data);
            list.add(s);
        }

        System.out.println("目前正在营业的商家" +list);
    }


    public static void main(String[] args) throws Exception {
        //每个手机或者应用，即用户
        Customers customers = new Customers();

        //zookeeper连接（用户打开美团app）
        customers.conn();

        //获取meituan下的所有子节点列表（获取商家列表）
        customers.getShopList();

        //业务逻辑处理（对比商家，下单点餐）
        customers.business();

    }



    private void business() throws IOException {
        System.out.println("用户正在浏览商家");
        System.in.read();
    }
}

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

可以看出，Zookeeper的主要功能是对某些列表存放在服务器中的处理，如相当于将集合作为一个服务器来处理，这样可以不用考虑很多数据的界限，这个集合通常代表某些元数据信息（由于是这样，所以他只是考虑数据不大的集合情况），并且可以考虑服务器的任何处理（如锁，分布式），所以如果可以，你也能手动来一个程序，手动的处理创建一个集合来完成与zookeeper相同的操作，只不过大多数手动的没有zookeeper这样的细节而已

从上面可以看出，节点的确是商家的信息，或者说节点就是商家，一般我们会在该商家，也就是节点里，加上数据

然后客户端，就得到这些商家，而他的数据（这里并没有操作，大概与订单有关），后面再进行操作吧

运行客户类，就会得到商家列表（商家服务类先不操作，先手动的添加对应节点，也就是商家）

首先在linux中添加一个商家，然后启动客户端，观察客户端的控制台输出（商家列表会立刻更新出最新商家信息）

多添加几个，启动后也会输出商家列表对应的数据

create /meituan/KFC "KFC" #先存放商家名称
create /meituan/BKC "BurgerKing"
create /meituan/baozi "baozi"
1
2
3

在linux中删除商家，再次启动客户端的控制台也会实时看到商家移除后的最新商家列表数据

delete /meituan/baozi
1

但是发现，程序在启动时，操作节点，并不会动态获取节点数据变化，必须要重新启动才会进行获取最新节点数据

即只显示了一次，那么这时我们就需要进行监听的操作了（上面并没有操作监听）

在客户类里修改部分代码如下：

//连接zookeeper
    public void conn() throws  Exception{
        zooKeeper = new ZooKeeper(connStr, session, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("监听的反馈");
                //再次获取商家列表
                try {
                    getShopList(); //内部类抛不出异常，类之间不可，即需要try-catch
                   //而正是因为程序没有执行完，被System.in.read()进行拦截，使得不会进行关闭，也就是说
                    //不会进行监听的删除关闭或者取消监听，即这里就形成了闭环（无限的循环监听）
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

运行商家服务类（以main方法带参数的形式运行）：

一些高版本的idea需要设置多次main方法，需要点击如下：

这里选中的地方，英文意思：允许多个实例

当设置了这个时，对应的main方法，就不在只操作一个线程了，而是多个线程，即可以运行多个main方法

接下来可以启动商家服务类和客户类来进行操作了（自己进行测试）

参数就可以当成该商家的名称（即这就是节点存放的数据），而创建的节点可以当成商家

由于商家是一个节点，那么他肯定有很多子节点的，且也有对应数据，即一个商家的信息是可以进行扩展的，这里注意一下

最后的细节：一般直接在java的main方法那里运行时，有默认的操作，这时不会显示给我们的

除非我们进行修改，才会在设置里显示，且我们添加时，一般需要进行该名，才会使得main的方法出现你修改的哪个配置

当然，是根据先后的，即后面的改变没有作用，当没有满足上面的操作时，就是要main的默认方式

而我们关闭窗口，其实未必只能点击关闭按钮来直接关闭，也可以按下esc按键来关闭

接下来，我来说明一些临时节点的细节（因为这里的商家服务类操作的就是临时节点，所以说明一下）：

前面说过，当对应的客户端关闭时，他的对应创建的临时节点会进行自动删除

所以当你关闭商家服务类时，对应应该还会有一个监听反馈，这时你可能看到并没有反馈

这是因为实际上自动删除这个节点需要时间，在java程序上需要很久，在zookeeper里不需要很久，甚至是非常快

所以当退出对应的zookeeper时，其他的zookeeper服务器可以及时的看到对应的节点删除，即发现删除了

而java程序却需要等待一些时间，才会看到节点删除，这是因为zookeeper客户端是内部的

而我们java程序不是，即需要更多的时间，主要是进行连接的操作（确定是否是这个客户端），耗费时间多

所以你可以等待一些时间，到那时你会发现，对应的反馈就出来了

当然了无论是自己写的客户端，还是zookeeper的客户端

对应的监听后的显示也只是显示在当前客户端

即监听显示并不会共享（不会在其他zookeeper服务器里显示，就如上面的代码一样，只会在当前程序里显示）

只是共享节点而已，但其他zookeeper服务器使得节点改变

还是会使得添加了监听的zookeeper服务器出现监听显示的，这是肯定的

案例-分布式锁-商品秒杀：

锁：我们在多线程中接触过，作用就是让当前的资源不会被其他线程访问，因为若都访问，在一些情况下会出现数据的不合理

因为数据的获取和修改不是立即的，而不加锁时，可以利用中间操作时间，而形成数据的不合理

而锁直接在源头，也就是分配的地方进行操作（判断是否有锁），使得不让你继续进行

若没锁，分配一个线程，即给对应的参数加上对应标志，实际上加锁也就是给对应标志

而正是由于分配，使得单个程序执行（正好抢占，抢占锁得到或者说再次得到资源）

单核来说，多核的话，会进行核的选中，然后在这个核里面进行抢占选择

而这个源头是一个服务器里面的或者说一个JVM进程里面的，因为就算再怎么分配，操作的参数也只是该JVM里面

即这个参数有有对应标志，所以不同服务器是操作不了的，即我们需要分布式锁操作不同JVM进程或者说服务器

实际上就是统一将这个标志用来给同一个地方，当成加锁，而这个标志的地方，就是分布式锁的核心

我的日记本，不可以被别人看到。所以要锁在保险柜中

当我打开锁，将日记本拿走了，别人才能使用这个保险柜

在zookeeper中使用传统的锁（有对应的统一地方放标志）引发的 “羊群效应” ：

1000个人创建节点，只有一个人能成功，999人需要等待！

羊群是一种很散乱的组织，平时在一起也是盲目地左冲右撞，但一旦有一只头羊动起来，其他的羊

也会不假思索地一哄而上，全然不顾旁边可能有的狼和不远处更好的草，羊群效应就是比喻人都有一种从众心理

从众心理很容易导致盲从，而盲从往往会陷入骗局或遭到失败

发现，用户对应的创建节点基本都是等待对应是否删除或者没有创建（一般都是等待删除），当知道删除或没有创建后

用户进行创建节点，否则一直等待，而不创建，这样后面的所有用户都需要等待，这样性能是很差的（将锁放在创建这里）

其中我们可以发现，对应的标志，应该就是这个临时节点了，那么我们就会去这个节点里进行查看，当然，与JVM的锁一样

是根据分配的（一般会操作范围的线程，即当一个线程被睡眠时，一般参与不了）

但是同样的与JVM锁一样，都是抢占（线程抢占和判断核，因为客户端连接服务器，一定是由线程来进行处理的）

这样就会出现一个问题，因为都是抢占，那么是没有顺序可说的（羊群效应），那么如何使得有顺序呢

避免"羊群效应"，zookeeper采用分布式锁（采用监听的操作，实现分布式锁的作用）

原来是根据临时节点的标志，来进行锁的作用，而现在在是根据监听通知的操作，来进行锁的作用

正是因为这样，所有就没有抢占一说了，从而实现了顺序的执行（无通知不执行）

但无论是什么锁的操作，都是进行一个进，其他人不进的理念

所有请求进来，在/lock下创建 临时顺序节点 （用来进行监听通知的，没有抢占的操作了，即可以全部请求进来）

放心，zookeeper会帮你编号排序

判断自己是不是/lock下最小的节点

是，获得锁（创建节点）

否，对前面小我一级的节点进行监听

获得锁请求，处理完业务逻辑，释放锁（删除节点），后一个节点有监听的情况下得到通知（比你年轻的死了，你成为最嫩的了）

然后再次进行判断是否是最小的节点，以此类推，而无监听的，则直接判断，无需等待通知

我们可以发现，这里直接创建所有的节点，而不是一个一个来，实际上只是锁住了业务操作

实现步骤：

初始化数据库：

创建数据库zkproduct，使用默认的字符集utf8

-- 创建数据库zkproduct，使用字符集utf8
CREATE DATABASE zkproduct CHARACTER SET utf8;

USE zkproduct;

-- 商品表
CREATE TABLE product(
 id INT PRIMARY KEY AUTO_INCREMENT, -- 商品编号
 product_name VARCHAR(20) NOT NULL, -- 商品名称
 stock INT NOT NULL, -- 库存
 VERSION INT NOT NULL -- 版本
);

INSERT INTO product (product_name,stock,VERSION) VALUES('锦鲤-清空购物车-大奖',5,0);

-- 订单表
CREATE TABLE `order`(
 id VARCHAR(100) PRIMARY KEY, -- 订单编号
 pid INT NOT NULL, -- 商品编号
 userid INT NOT NULL -- 用户编号
);
-- 若出现错误，那么删除掉注释，因为可能有隐藏的符号（一般是没有的）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

搭建工程：

搭建ssm框架，操作对库存表-1，对订单表+1

对应的具体工程（也可以不与这个相同，只要作用一样即可）：

对应的依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-
                             4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.lagou</groupId>
    <artifactId>zk_product</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <spring.version>5.2.7.RELEASE</spring.version>
    </properties>

    <packaging>war</packaging>

    <dependencies>
     <!-- Spring -->
     <dependency>
         <!--IOC容器的对象需要-->
       <groupId>org.springframework</groupId>
       <artifactId>spring-context</artifactId>
       <version>${spring.version}</version>
     </dependency>
     <dependency>
         <!--基础依赖，其他依赖一般会导入这个，这里进行版本操作一下，实际上可以不写-->
       <groupId>org.springframework</groupId>
       <artifactId>spring-beans</artifactId>
       <version>${spring.version}</version>
     </dependency>
     <dependency>
         <!--有对应的类操作页面，如前端控制器-->
       <groupId>org.springframework</groupId>
       <artifactId>spring-webmvc</artifactId>
       <version>${spring.version}</version>
     </dependency>
     <dependency>
         <!--spring使用连接池的，包括一些tx操作
		当需要tx的一些操作时（如事务传播），那么可以导入tx-->
       <groupId>org.springframework</groupId>
       <artifactId>spring-jdbc</artifactId>
       <version>${spring.version}</version>
     </dependency>
     <!-- Mybatis -->
     <dependency>
         <!--引入mybatis依赖，如工厂的一些类-->
       <groupId>org.mybatis</groupId>
       <artifactId>mybatis</artifactId>
       <version>3.5.5</version>
     </dependency>
     <dependency>
         <!--可以使用注解，配置mybatis-->
       <groupId>org.mybatis</groupId>
       <artifactId>mybatis-spring</artifactId>
       <version>2.0.5</version>
     </dependency>
     <!-- 连接池 -->
     <dependency>
       <groupId>com.alibaba</groupId>
       <artifactId>druid</artifactId>
       <version>1.1.10</version>
     </dependency>
     <!-- 数据库驱动 -->
     <dependency>
       <groupId>mysql</groupId>
       <artifactId>mysql-connector-java</artifactId>
       <version>8.0.20</version>
     </dependency>
     <!-- junit @Test的操作-->
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <version>4.12</version>
       <scope>test</scope>
     </dependency>
    </dependencies>
    <build>
         <plugins>
           <!-- maven内嵌的tomcat插件 -->
           <plugin>
             <groupId>org.apache.tomcat.maven</groupId>
             <!-- 目前apache只提供了tomcat6和tomcat7两个插件 -->
             <artifactId>tomcat7-maven-plugin</artifactId>
               <version>2.1</version>
             <configuration>
               <port>8001</port>
               <path>/</path>
             </configuration>
             <executions>
               <execution>
                 <!-- 打包完成后,运行服务 -->
                 <phase>package</phase>
                 <goals>
                   <goal>run</goal>
                 </goals>
               </execution>
             </executions>
           </plugin>
             <plugin>
             <!--maven的插件依赖，这里是对应编译插件，设置参数-->
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
                 <version>3.1</version>
                    <configuration>
             <!--编译级别，指定JDK的版本以及操作编码，因为编译也要得到数据，那么也就需要编码-->
                      <source>11</source>
                      <target>11</target>
                        <!--在设置里的Java Compiler，可以看到对应的版本发生了变化，记得刷新-->
                      <encoding>UTF-8</encoding>
                    </configuration>
                  </plugin>
         </plugins>
    </build>

</project>
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

mybatis的配置文件（mybatis-config.xml）：

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE configuration PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>
     <!-- 后台的日志输出：针对开发者-->
     <settings>
       <setting name="logImpl" value="STDOUT_LOGGING"/>
     </settings>
</configuration>

1
2
3
4
5
6
7
8
9
10

spring的配置文件（spring.xml）：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:mvc="http://www.springframework.org/schema/mvc"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:tx="http://www.springframework.org/schema/tx"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
           http://www.springframework.org/schema/beans/spring-beans.xsd
           http://www.springframework.org/schema/context
           http://www.springframework.org/schema/context/spring-context.xsd
           http://www.springframework.org/schema/tx
           http://www.springframework.org/schema/tx/spring-tx.xsd">
    <!-- 1.扫描包下的注解 -->
    <context:component-scan base-package="controller,service,mapper"/>
    <!-- 2.创建数据连接池对象 -->
    <bean id="dataSource" class="com.alibaba.druid.pool.DruidDataSource"
          destroy-method="close">
        <!--destroy-method="close"的作用是当数据库连接不使用的时候
		就把该连接重新放到数据池中，方便下次使用调用，实际上就是调用close方法
在最后销毁的时候，而由于连接池的close方法就是归还连接，所以就算重新放入连接池中
因为这里只是提供连接池，并没有操作最终的练级去向，即还是需要我们进行关闭
-->
        <!--serverTimezone是数据库连接中的参数，用于设置服务时间
		标识设置服务时间为东一区时间，即国际日期变更线时间-->
        <property name="url" value="jdbc:mysql://192.168.164.128:3306/zkproduct?serverTimezone=GMT"/>
        <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
        <property name="username" value="root"/>
        <property name="password" value="QiDian@666"/>

    </bean>
    <!-- 3.创建SqlSessionFactory，并引入数据源对象 -->
    <bean id="sqlSessionFactory"
          class="org.mybatis.spring.SqlSessionFactoryBean">
        <property name="dataSource" ref="dataSource"></property>
        <property name="configLocation" value="classpath:mybatis/mybatis-config.xml"></property>

    </bean>
    <!-- 4.告诉spring容器，数据库语句代码在哪个文件中-->
    <!-- 如mapper.xDao接口对应resources/mapper/xDao.xml-->
    <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">
        <property name="basePackage" value="mapper"></property>
    </bean>
    <!-- 5.将数据源关联到事务 -->
    <bean id="transactionManager"
          class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
        <property name="dataSource" ref="dataSource"></property>
    </bean>
    <!-- 6.开启事务 -->
    <tx:annotation-driven/>
</beans>

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

web.xml配置文件：

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee
        http://xmlns.jcp.org/xml/ns/javaee/web-app_3_1.xsd"
         version="3.1">
    <servlet>
        <servlet-name>springMVC</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>

        <init-param>
            <param-name>contextConfigLocation</param-name>
            <param-value>classpath:spring/spring.xml</param-value>
        </init-param>
        <load-on-startup>1</load-on-startup>
        <async-supported>true</async-supported> 
        <!--作用是支持异步处理-->
        <!--一般的servlet由一个线程来操作，我们知道，只有当响应数据后，这个线程才基本结束
即会有页面的访问完成
若业务代码非常耗时，使得该线程资源一直占用，即页面一直没有访问完成
而使用异步处理，可以操作业务代码时，页面以及访问完成，只是需要等待一些时间，就与ajax类似
正是因为该线程是异步的，所以就出现了类似ajax的操作-->
    </servlet>
    <servlet-mapping>
        <servlet-name>springMVC</servlet-name>
        <url-pattern>/</url-pattern>
    </servlet-mapping>
</web-app>
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

对应的OrderMapper接口：

package mapper;

import models.Order;
import org.apache.ibatis.annotations.Insert;
import org.apache.ibatis.annotations.Mapper;
import org.springframework.stereotype.Component;

/**
 *
 */
@Mapper //扫描这里（使用了注解的扫描），使得mapper即可以说是注解会起作用
//一般操作注解，因为配置文件的操作，一般都不是他来扫描的
//mybatis启动时会找他，整个项目找，找到@Mapper后，扫描当前类
//那么对应的注解也会起作用
//一般要与spring整合时使用，创建当前实例放到ioc容器里面，然后注入使得调用方法（因为sql语句的注解起作用了）
//但是单独使用的话，基本不会有对应的作用
//而不用配置文件扫描了（虽然上面也扫描了，即对应配置文件已经扫描了，所以可以不写）
@Component 
//这个可以删掉，因为配置里面有这个实例，且他会完全覆盖这个其他实例，因为他是在扫描后进行覆盖
//所以这个就没有起作用，即没有相同的实例，所以注入可以操作

//上面两个都可以不写
public interface OrderMapper {
  // 生成订单
  @Insert("insert into `order` (id,pid,userid) values (#{id},#{pid},#{userid})")
  int insert(Order order);
    
    //注意：由于order在mysql里面是关键字，所有我们需要显示的指定，即加上``，来包括
}

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

对应的ProductMapper接口：

package mapper;

import models.Product;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.Update;
import org.springframework.stereotype.Component;

/**
 *
 */
@Mapper
@Component
public interface ProductMapper {
    // 查询商品（目的查库存）
  @Select("select * from product where id = #{id}")
  Product getProduct(@Param("id") int id);
  
  // 减库存
  @Update("update product set stock = stock-1 where id = #{id}")
  int reduceStock(@Param("id") int id);
}

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

对应的Order类：

package models;

import java.io.Serializable;

/**
 *
 */
public class Order implements Serializable {

    private String id;
    private int pid;
    private int userid;

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public int getPid() {
        return pid;
    }

    public void setPid(int pid) {
        this.pid = pid;
    }

    public int getUserid() {
        return userid;
    }

    public void setUserid(int userid) {
        this.userid = userid;
    }

    public Order() {
    }

    public Order(String id, int pid, int userid) {
        this.id = id;
        this.pid = pid;
        this.userid = userid;
    }
}
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

对应的Product类：

package models;

import java.io.Serializable;

/**
 *
 */
public class Product implements Serializable {
    private int id;
    private String product_name;
    private int stock;
    private int version;

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getProduct_name() {
        return product_name;
    }

    public void setProduct_name(String product_name) {
        this.product_name = product_name;
    }

    public int getStock() {
        return stock;
    }

    public void setStock(int stock) {
        this.stock = stock;
    }

    public int getVersion() {
        return version;
    }

    public void setVersion(int version) {
        this.version = version;
    }

    public Product() {
    }

    public Product(int id, String product_name, int stock, int version) {
        this.id = id;
        this.product_name = product_name;
        this.stock = stock;
        this.version = version;
    }
}

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

对应的ProductService类及其实现类

只需要写这一个类，看看对应实现类的操作就可以知道为什么，因为操作了多个方法，这就是一个业务，即业务层的操作：

package service;



/**
 *
 */
public interface ProductService {

    // 减库存
    int reduceStock(int id);
}

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

package service.impl;

import mapper.OrderMapper;
import mapper.ProductMapper;
import models.Order;
import models.Product;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import service.ProductService;

import java.util.UUID;

/**
 *
 */
@Service
public class ProductServiceImpl implements ProductService {

    @Autowired
    private ProductMapper productMapper;

    @Autowired
    private OrderMapper orderMapper;

    // 减库存
    @Override
    public int reduceStock(int id) {
        //获取库存（根据商品id查询商品）
        Product product = productMapper.getProduct(id);
        if (product.getStock() <= 0)
            throw new RuntimeException("已抢光！");
        //减库存
        int i = productMapper.reduceStock(id);
        if (i == 1) {
            //生成订单
            Order order = new Order();
            order.setId(UUID.randomUUID().toString()); //使用UUID工具帮我们生成一个订单号
            order.setPid(id);
            order.setUserid(101);
            orderMapper.insert(order);
        } else {
            throw new RuntimeException("减库存失败，请重试！");

        }
        return i;
    }

}

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

对应的ProductAction类：

package controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import service.ProductService;

/**
 *
 */
@Controller
public class ProductAction {
    @Autowired
    private ProductService productService;

    @GetMapping("/product/reduce")
    @ResponseBody
    public Object reduceStock(int id) throws Exception {
        productService.reduceStock(id);
        return "ok";
    }
}


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

启动测试 ：

点击打包按钮，如图：

接下来访问http://localhost:8001/product/reduce?id=1，可以发现，数据库的数据发生改变了

现在我们进行下面的操作

启动两次工程，端口号分别8001和8002（相当于两个服务器）

使用nginx做负载均衡

在这之前，说明一些主机域名localhost和127.0.0.1，他们为什么可以不使用网络来进行本机的访问

一般情况下，我们需要发送分组，在不是本机的访问时，需要进行网络才能有分组过去，而本机却有一个专门处理分组的地方

使得与本机与应用程序连接（端口之间发生分组）

当然linux也是一样的，当访问本机时，可以不用执行对应ip地址（我们设置的）

也可直接的操作localhost或者代表本机的ip地址（如127.0.0.1）

使得本机端口进行通信（有些情况下，对应ip只要127开头就可以了）

启动nginx可以用如下命令：

./nginx -c conf/nginx.conf 
#后面默认在nginx目录里，即指定该配置文件启动，若你使用/开头，那么就是根目录开始，否则默认nginx目录开始
1
2

upstream sga{
		server 192.168.164.1:8001;
		server 192.168.164.1:8002;  #根据cmd的ipconfig，查看本机的ip地址，都是网卡，基本随便一个都可
	}
	server {
		listen  80;
		#server_name localhost; #localhost就算设置域名，也是默认本机，即127.0.0.1
 		server_name 192.168.164.128; #即这里改一下
	location / {
		proxy_pass http://sga;
		root  html;
		index ?index.html index.htm;
	}
  
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

使用 JMeter 模拟1秒内发出10个http请求

先下载JMeter：

下载地址：http://jmeter.apache.org/download_jmeter.cgi（最好选择直接看到的最新的上面的zip文件）

解压后，在bin目录下点击如下：

即可打开，操作如下：

先添加线程组（Thread Group）

设置如下：

再在该线程组上添加HTTP请求（HTTP Request）

设置如下：

后面的参数那里，记得进行添加列，使得可以加锁参数，要不然一般操作不了，因为没有点击的地方

再在HTTP请求（HTTP Request）上添加结果树视图（View Results Tree）

可以看到我们的请求结果是否成功

点击如下：

上面不同背景的，就是要点击的

查看测试结果，若是正常的，可以多试几次（如将库存改成10，请求100次）

查看数据库，若stock库存变成负数，则是并发导致的数据结果错误（不同的项目操作）

正好同时得到，即同时进行操作，先判断完毕了

若出现这样的情况，在实际生活中，基本会造成大量的亏损，所有必须避免，因为订单是实际存在的，用户是需要得到该奖品的

除非你不想办了

上面只是在介绍分布式锁之前，进行的小测试，接下来看看

如何使用分布式锁解决（传统的锁操作也可以，只是分布式锁操作在传统的锁操作上进行了顺序操作），临时节点当作标记

而JVM的锁不可以，因为对应的锁监听就算是设置static也不同的，或者说对应的指向完全不同

又或者说JVM的空间，实际上就是标记不会被看到，因为在不同的项目部署下

所有你会发现，就算设置了synchronized锁，也是没有用的

apahce提供的zookeeper客户端：

基于zookeeper原生态的客户端类实现分布式是非常麻烦的，我们使用apahce提供了一个zookeeper客户端来实现

官网：http://curator.apache.org/

对应依赖（记得加上，刷新）：

<dependency>
    <!--有对应的类，如CuratorFramework操作（curator工具对象）-->
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>4.2.0</version> <!-- 网友投票最牛逼版本 -->
</dependency>
1
2
3
4
5
6

recipes是curator族谱大全，里面包含zookeeper和framework

在控制层中加入分布式锁的逻辑代码（对ProductAction类的修改）：

package controller;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import service.ProductService;

/**
 *
 */
@Controller
public class ProductAction {
    @Autowired
    private ProductService productService;

    private static String connectString = 
        "192.168.164.128:2181,192.168.164.129:2181,192.168.164.130:2181";


    @GetMapping("/product/reduce")
    @ResponseBody
    public Object reduceStock(int id) throws Exception {


        //重试策略（1000毫秒试一次，最多试3次），进行连接集群的用处，若还是连不上集群，就会报错
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

        //创建curator工具对象
        CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);

        //启动
        client.start();

        //根据工具对象，创建内部互斥锁
        InterProcessMutex lock = new InterProcessMutex(client, "/product_" + id); 
        //后面会帮我们创建编号，前面说过的

        try {
            //加锁
            lock.acquire();
            productService.reduceStock(id);
        }catch (Exception e){
            e.printStackTrace();
            if(e instanceof RuntimeException){
                throw e;
            }
        }finally {
            //释放锁
            lock.release(); //会删除当前节点，并有通知，可以看到，这个节点差不多是可以随便写的
            //不管你是否操作成功，都要打开锁
            
            //可能会有疑问，假设释放锁后，再进行请求会怎么样，前面说过
            //有监听的，那么等待通知，然后判断释放最小节点，无监听的直接判断最小节点
            //而这样的，相当于没有监听的，那么直接判断是否是最小节点
        }
        return "ok";
    }
}


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

再次测试，发现，无论测试多少次，基本都没有负数的库存，即并发问题解决

分布式锁的原理只不过是利用所有请求在服务器或者集群服务器中的锁来处理而已，简单来说，就是锁的对应资源保证一致