07年今年多大岁了(0751是哪个城市的区号)

分布式技术-Zookeeper

Zookeeper概述Zookeeper本地模式安装Zookeeper内部原理Zookeeper实战

1. Zookeeper本地模式安装 1.1 概述

美团，饿了么，淘宝，58同城等等应用都是zookeeper的现实生活版我开了个饭店，如何才能让大家都能吃到我们的饭菜？需要入驻美团，这样大家就可以在美团app中看到我的饭店，下订单，从而完成一次交易Zookeeper是一个开源的分布式（多台服务器干一件事）的，为分布式应用提供协调服务的 Apache项目。在大数据技术生态圈中，zookeeper（动物管理员），Hadoop（大象），Hive（蜜蜂）， Pig（猪）等技术以动物相关为名称。 1.2 工作机制

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

它负责 存储 和 管理 大家都关心的数据

然后接受观察者的注册，一旦这些数据的发生变化Zookeeper就将负责通知已经注册的那些观察者做出相应的反应从而实现集群中类似Master/Slave管理模式

Zookeeper = 文件系统 + 通知机制

商家营业并入驻获取到当前营业的饭店列表服务器节点下线服务器节点上下线事件通知重新再去获取服务器列表，并注册监听

1.3 特点

分布式和集群的区别？

无论分布式和集群，都是很多人在做事情。具体区别如下：
例如：我有一个饭店，越来越火爆，得多招聘一些工作人员

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

是一个leader和多个follower来组成的集群（狮群中，一头雄狮，N头母狮）集群中只要有半数以上的节点存活，Zookeeper就能正常工作（5台服务器挂2台，没问题；4台服务器挂2台，就停止）全局数据一致性，每台服务器都保存一份相同的数据副本，无论client连接哪台server，数据都是一致的数据更新原子性，一次数据要么成功，要么失败（不成功便成仁）实时性，在一定时间范围内，client能读取到最新数据更新的请求按照顺序执行，会按照发送过来的顺序，逐一执行（发来123，执行123，而不是321或者别的）

1.4 数据结构

ZooKeeper数据模型的结构与linux文件系统很类似，整体上可以看作是一棵树，每个节点称做一 个ZNode（ZookeeperNode）。

每一个ZNode默认能够存储1MB的数据（元数据），每个ZNode的路径都是唯一的

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

1.5 应用场景

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

1.5.1 统一命名服务

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

例如：服务器的IP地址不容易记，但域名相比之下却是很容易记住

1.5.2 统一配置管理

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

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

将配置管理交给Zookeeper
1、将配置信息写入到Zookeeper的某个节点上
2、每个客户端都监听这个节点
3、一旦节点中的数据文件被修改，Zookeeper这个话匣子就会通知每台客户端服务器 1.5.3 服务器节点动态上下线

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

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

1.5.4 软负载均衡

Zookeeper会记录每台服务器的访问数，让访问数最少的服务器去处理最新的客户请求（雨露均沾）都是自己的孩子，得一碗水端平
1.6 下载地址

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

apache-zookeeper-3.6.0.tar.gz需要安装maven，然后再运行mvn clean install 和mvn javadoc:aggregate，前一个命令会下载安装好多jar包，不知道要花多长时间apache-zookeeper-3.6.0-bin.tar.gz已经自带所需要的各种jar包，所以下载bin.tar.gz为后缀的版本！ 2. Zookeeper本地模式安装 2.1 本地模式安装 2.1.1 安装前准备

安装jdk拷贝apache-zookeeper-3.6.0-bin.tar.gz到opt目录解压安装包、重命名

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

2.1.2 配置修改

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

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

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

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

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

dataDir=/opt/zookeeper/zkData
dataLogDir=/opt/zookeeper/zkLog

操作Zookeeper（启动、查看状态、退出）

在bin目录下启动Zookeeper

[root@localhost bin]# ./zkServer.sh start
启动成功：Starting zookeeper ... STARTED

查看进程是否启动

[root@localhost bin]# jps
# QuorumPeerMain：是zookeeper集群的启动入口类，是用来加载配置启动QuorumPeer线程的
#当显示：22131 QuorumPeerMain 

查看状态：

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

Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: standalone

当发现报错：Error contacting service. It is probably not running.
说明默认端口8080被占用了，那么就可以考虑更换Zookeeper的端口
操作方式：https://blog.csdn.net/qq_21137441/article/details/102662839
对zoo.cfg中的值进行添加或修改：admin.serverPort=9099
4. 启动客户端

[root@localhost bin]# ./zkCli.sh
#查看当前的进程 [zk: localhost:2181(CONNECTED) 0] ls /

退出客户端

[zk: localhost:2181(CONNECTED) 0] quit

2.2 配置参数解读

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

tickTime =2000 ：通信心跳数，Zookeeper服务器与客户端心跳时间，单位毫秒

Zookeeper使用的基本时间，服务器之间或客户端与服务器之间维持心跳的时间间隔，也就 是每个tickTime时间就会发送一个心跳，时间单位为毫秒。

initLimit =10 ：LF(Leader,Follower)初始通信时限

集群中的Follower跟随者服务器与Leader领导者服务器之间，启动时能容忍的最多心跳数;10*2000（10个心跳时间）如果领导和跟随者没有发出心跳通信，就视为失效的连接，领导和跟随者彻底断开

syncLimit =5 ：LF同步通信时限

集群启动后，Leader与Follower之间的最大响应时间单位，假如响应超过syncLimit * tickTime->10秒，Leader就认为Follwer已经死掉，会将Follwer从服务器列表中删除

dataDir ：数据文件目录+数据持久化路径，主要用于保存Zookeeper中的数据

dataLogDir ：日志文件目录

clientPort =2181：客户端连接端口（监听客户端连接的端口。）

3. Zookeeper内部原理 3.1 选举机制（面试重点）

半数机制：集群中半数以上机器存活，集群可用。所以Zookeeper适合安装奇数台服务器虽然在配置文件中并没有指定Master和Slave。但是，Zookeeper工作时，是有一个节点为 Leader，其他则为Follower，Leader是通过内部的选举机制临时产生的

Server1先投票，投给自己，自己为1票，没有超过半数，根本无法成为leader，顺水推舟将票数投给了id比自己大的Server2Server2也把自己的票数投给了自己，再加上Server1给的票数，总票数为2票，没有超过半数，也无法成为leader，也学习Server1，顺水推舟，将自己所有的票数给了id比自己大的Server3Server3得到了Server1和Server2的两票，再加上自己投给自己的一票。3票超过半数，顺利成为leaderServer4和Server5都投给自己，但是无法改变Server3的票数，只好听天由命，承认Server3是leader

3.2 节点类型

持久型（persistent）：

持久化目录节点（persistent）客户端与zookeeper断开连接后，该节点依旧存在持久化顺序编号目录节点（persistent_sequential）客户端与zookeeper断开连接后，该节点依旧存在，创建znode时设置顺序标识，znode名称后会附加一个值，顺序号是一个单调 递增的计数器，由父节点维护，例如：Znode001，Znode002…

短暂型（ephemeral）：

临时目录节点（ephemeral）客户端和服务器端断开连接后，创建的节点自动删除临时顺序编号目录节点（ephemeral_sequential）客户端与zookeeper断开连接后，该节点 被删除，创建znode时设置顺序标识，znode名称后会附加一个值，顺序号是一个单调递增 的计数器，由父节点维护，例如：Znode001，Znode002…

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

3.3 监听器原理（面试重点）

在main方法中创建Zookeeper客户端的同时就会创建两个线程，一个负责网络连接通信，一个负责监听监听事件就会通过网络通信发送给zookeeperzookeeper获得注册的监听事件后，立刻将监听事件添加到监听列表里zookeeper监听到 数据变化 或 路径变化，就会将这个消息发送给监听线程
常见的监听:
1. 监听节点数据的变化：get path [watch]
2. 监听子节点增减的变化：ls path [watch]监听线程就会在内部调用process方法（需要我们实现process方法内容）

3.4 写数据流程

Client 想向 ZooKeeper 的 Server1 上写数据，必须的先发送一个写的请求如果Server1不是Leader，那么Server1 会把接收到的请求进一步转发给Leader。这个Leader 会将写请求广播给各个Server，各个Server写成功后就会通知Leader。当Leader收到半数以上的 Server 数据写成功了，那么就说明数据写成功了。随后，Leader会告诉Server1数据写成功了。Server1会反馈通知 Client 数据写成功了，整个流程结束

4. Zookeeper实战（开发重点） 4.1 分布式安装部署

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

4.1.1 安装zookeeper

参考本地模式安装2.1

4.1.2 配置服务器编号

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

[root@localhost zkData]# vim myid

在文件中添加与server对应的编号：1 （只要在myid文件中写入1即可）；其余两台服务器分别对应2和3 4.1.3 配置zoo.cfg文件

打开zoo.cfg文件，增加如下配置

#######################cluster##########################
server.1=192.168.212.128:2888:3888
server.2=192.168.212.129:2888:3888
server.3=192.168.212.130:2888:3888

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

A：一个数字，表示第几号服务器。集群模式下配置的/opt/zookeeper/zkData/myid文件里面的数据就是A的值B：服务器的ip地址C：与集群中Leader服务器交换信息的端口D：选举时专用端口，万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口。 4.1.4 配置其余两台服务器

在虚拟机数据目录vms下，创建zk02将本台服务器数据目录下的.vmx文件和所有的.vmdk文件分别拷贝zk02下虚拟机->文件->打开 （选择zk02下的.vmx文件）开启此虚拟机，弹出对话框，选择“我已复制该虚拟机”进入系统后，修改linux中的ip: https://blog.csdn.net/M983373615/article/details/100031125修改/opt/zookeeper/zkData/myid中的数值为2第三台服务器zk03，重复上面的步骤

4.1.5 集群操作

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

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

启动第1台

[root@localhost /]# cd opt/zookeeper/bin/
[root@localhost bin]# ./zkServer.sh start

查看状态

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

以下为状态信息：

ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Error contacting service. It is probably not running.

注意：
1.因为没有超过半数以上的服务器，所以集群失败 （防火墙没有关闭也会导致失败）
2.可能上面的防火墙关闭失败,关闭防火墙：

  #查看防火墙状态
   service iptables status 
  #关闭防火墙
   service iptables stop
  #查看防火墙开机启动状态
   chkconfig iptables --list
  #关闭防火墙开机启动
   chkconfig iptables off

当启动第2台服务器时

查看第1台的状态：Mode: follower查看第2台的状态：Mode: leader

4.2 客户端命令行操作 启动和查看节点

在1号操作

启动客户端

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

显示所有操作命令

help

查看当前znode中所包含的内容，斜杆/表示根目录

ls /

查看当前节点详细数据，zookeeper老版本使用 ls2 / ，现在已经被新命令替代

ls -s /

ls -s /查看当前节点详细数据讲解：

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
# 在根目录下，创建俄罗斯节点，并保存“普京”数据到节点上
create /ru "pujing"
# 多级创建节点：在日本下，创建东京 “热”，japan必须提前创建好，否则报错 “节点不存在”
create /japan/Tokyo "hot"

# 获得节点的值
get /japan/Tokyo

# 创建短暂节点：创建成功之后，quit退出客户端，重新连接，短暂的节点消失
create -e /uk
ls /
quit
ls /

# 创建带序号的节点，在俄罗斯ru下，创建3个city
create -s /ru/city # 执行三次
[zk: localhost:2181(CONNECTED) 1] create -s /ru/city
Created /ru/city0000000000
[zk: localhost:2181(CONNECTED) 2] create -s /ru/city
Created /ru/city0000000001
[zk: localhost:2181(CONNECTED) 3] create -s /ru/city
Created /ru/city0000000002

ls /ru

# 输出：[city0000000000, city0000000001, city0000000002]
# 如果原来没有序号节点，序号从0开始递增。如果原节点下已有2个节点，则再排序时从2开始，以此类推

# 修改节点数据值
set /japan/Tokyo "too hot"

# 删除节点
delete /usa/NewYork
# 递归删除节点 （非空节点，节点下有子节点）
# 不仅删除/ru，而且/ru下的所有子节点也随之删除
deleteall /ru

监听

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

# 1. 在server3主机上注册监听/usa节点的数据变化
addWatch /usa

# 2. 在Server1主机上修改/usa的数据
set /usa "telangpu"
# Server3会立刻响应：
# WatchedEvent state:SyncConnected type:NodeDataChanged path:/usa

# 3. 如果在Server1的/usa下面创建子节点NewYork
create /usa/NewYork
# Server3会立刻响应：
# WatchedEvent state:SyncConnected type:NodeCreatedpath:/usa/NewYork

4.3 API应用 4.3.1 IDEA环境搭建

创建一个Maven工程添加pom文件

    
        org.apache.logging.log4j
        log4j-core
        2.8.2
    
    
        org.apache.zookeeper
        zookeeper
        3.6.0
    
    
        junit
        junit
        4.12
    

在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/zk.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

4.3.2 创建ZooKeeper客户端

test.TestZK

public class TestZK {
    // zookee[er集群ip和端口
    private String connString = "192.168.212.128:2181,192.168.212.129:2181,192.168.212.130:2181";
    
    private int sessionTimeout = 90000;
	//zookeeper客户端对象
    private ZooKeeper zkClient;
    @Test
    public void init() throws IOException {
        // 创建监听器
        Watcher watcher = new Watcher() {
            public void process(WatchedEvent watchedEvent) {
            }
        };
        // 创建zookeeper客户端
        zk = new ZooKeeper(connString, sessionTimeout, watcher);
    }
}

4.3.3 创建节点

一个ACL对象就是一个Id和permission对

表示哪个/哪些范围的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：创建者才有全部权限

@Before
public void init() throws IOException{
    // 省略...
}
@Test
public void createNode() throws Exception {
    String nodeCreated = zKcli.create("/lagou", "laosun".getBytes(),
                                      ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    // 参数1：要创建的节点的路径
    // 参数2：节点数据
    // 参数3：节点权限
    // 参数4：节点的类型
    System.out.println("nodeCreated = " + nodeCreated);
}

4.3.4 查询节点的值

@Test
public void find() throws Exception{
    byte[] bs = zKcli.getData("/lagou", false, new Stat()); // 路径不存在时会报错
    String data = new String(bs);
    System.out.println("查询到数据："+data);
}

4.3.5 修改节点的值

@Test
public void update()throws Exception{
    //先查看节点详情，获dataVersion = 0
    Stat stat = zKcli.setData("/lagou", "laosunA".getBytes(), 0); 
    System.out.println(stat);
}

4.3.6 删除节点

@Test
public void delete() throws Exception {
    zKcli.delete("/lagou", 1); // 先查看节点详情，获得dataVersion = 1
    System.out.println("删除成功！");
}

4.3.7 获取子节点

@Test
public void getChildren() throws Exception {
    List children = zKcli.getChildren("/",false); // false:不监听
    for (String child : children) {
        System.out.println(child);
    }
}

4.3.8 监听子节点的变化

@Before
public void init() throws Exception {
    zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        public void process(WatchedEvent watchedEvent) {
            System.out.println("得到监听反馈，再进行的业务处理代码！");
            System.out.println(watchedEvent.getType());
        }
    });
}

@Test
public void watchNode() throws Exception {
    List children = zKcli.getChildren("/", true); // true：注册监听
    for (String child : children) {
        System.out.println(child);
    }
    // 让线程不停止，等待监听的响应
    System.in.read();
}

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

IDEA的控制台就会做出响应：NodeChildrenChanged–/

4.3.9 判断Znode是否存在

@Test
public void exist() throws Exception {
    Stat stat = zKcli.exists("/lagou", false);
    System.out.println(stat == null ? "不存在" : "存在");
}

4.4 案例-模拟美团商家上下线 4.4.1 需求

模拟美团服务平台，商家营业通知，商家打烊通知
提前在根节点下，创建好 /meituan 节点

4.4.2 商家服务类

public class ShopServer {
    private static String connectString =
        "192.168.204.141:2181,192.168.204.142:2181,192.168.204.143:2181";
    private static int sessionTimeout = 60000;
    private ZooKeeper zk = null;
    // 创建到zk的客户端连接
    public void getConnect() throws IOException {
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            public void process(WatchedEvent event) {
            }
        });
    }
    // 注册到集群
    public void register(String ShopName) throws Exception {
        // 一定是"EPHEMERAL_SEQUENTIAL短暂有序型"的节点，才能给shop编号，shop1，
        shop2...”
            String create = zk.create("/meituan/Shop", ShopName.getBytes(),
                                      ZooDefs.Ids.OPEN_ACL_UNSAFE, 
                                      CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println("【"+ShopName+"】 开始营业！ " + create);
    }
    // 业务功能
    public void business(String ShopName) throws Exception {
        System.out.println("【"+ShopName+"】 正在营业中 ...");
        System.in.read();
    }
    public static void main(String[] args) throws Exception {
        ShopServer shop = new ShopServer();
        // 1.连接zookeeper集群（和美团取得联系）
        shop.getConnect();
        // 2.将服务器节点注册（入住美团）
        shop.register(args[0]);
        // 3.业务逻辑处理（做生意）
        shop.business(args[0]);
    }
}

4.4.3 客户类

public class Customers {
    private static String connectString =
        "192.168.204.141:2181,192.168.204.142:2181,192.168.204.143:2181";
    private static int sessionTimeout = 60000;
    private ZooKeeper zk = null;
    // 创建到zk的客户端连接
    public void getConnect() throws IOException {
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            public void process(WatchedEvent event) {
                // 再次获取所有商家
                try {
                    getShopList();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
    // 获取服务器列表信息
    public void getShopList() throws Exception {
        // 1获取服务器子节点信息，并且对父节点进行监听
        List shops = zk.getChildren("/meituan", true);
        // 2存储服务器信息列表
        ArrayList shoplist = new ArrayList();
        // 3遍历所有节点，获取节点中的主机名称信息
        for (String shop : shops) {
            byte[] data = zk.getData("/meituan/" + shop, false, new Stat());
            shoplist.add(new String(data));
        }
        // 4打印服务器列表信息
        System.out.println(shoplist);
    }
    // 业务功能
    public void business() throws Exception {
        System.out.println("客户正在浏览商家 ...");
        System.in.read();
    }
    public static void main(String[] args) throws Exception {
        // 1.获取zk连接 （客户打开美团）
        Customers client = new Customers();
        client.getConnect();
        // 2.获取/meituan的子节点信息，从中获取服务器信息列表（从美团中获取商家列表）
        client.getShopList();
        // 3.业务进程启动 （对比商家，点餐）
        client.business();
    }
}

运行客户类，就会得到商家列表首先在linux中添加一个商家，然后观察客户端的控制台输出（商家列表会立刻更新出最新商
家），多添加几个，也会实时输出商家列表

create /meituan/KFC "KFC"
create /meituan/BKC "BurgerKing"
create /meituan/baozi "baozi"

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

delete /meituan/baozi

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

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

锁：我们在多线程中接触过，作用就是让当前的资源不会被其他线程访问！
我的日记本，不可以被别人看到。所以要锁在保险柜中
当我打开锁，将日记本拿走了，别人才能使用这个保险柜

在zookeeper中使用传统的锁引发的 “羊群效应” ：1000个人创建节点，只有一个人能成功，999 人需要等待！

羊群是一种很散乱的组织，平时在一起也是盲目地左冲右撞，但一旦有一只头羊动起来，其他的羊 也会不假思索地一哄而上，全然不顾旁边可能有的狼和不远处更好的草。羊群效应就是比喻人都有 一种从众心理，从众心理很容易导致盲从，而盲从往往会陷入骗局或遭到失败。

避免“羊群效应”，zookeeper采用分布式锁

所有请求进来，在/lock下创建 临时顺序节点 ，放心，zookeeper会帮你编号排序（序号从0开始递增）

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

是，获得锁（创建节点）；否，对前面小我一级的节点进行监听

获得锁请求，处理完业务逻辑，释放锁（删除节点），后一个节点得到通知

重复步骤2

实现步骤 1. 初始化数据库

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

-- 商品表
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 -- 用户编号
)

2. 搭建工程

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

pom

war

    5.2.7.RELEASE


    
    
        org.springframework
        spring-context
        ${spring.version}
    
    
        org.springframework
        spring-beans
        ${spring.version}
    
    
        org.springframework
        spring-webmvc
        ${spring.version}
    
    
        org.springframework
        spring-jdbc
        ${spring.version}
    
    
    
        org.mybatis
        mybatis
        3.5.5
    
    
        org.mybatis
        mybatis-spring
        2.0.5
    
    
    
        com.alibaba
        druid
        1.1.10
    
    
    
        mysql
        mysql-connector-java
        8.0.20
    
    
    
        junit
        junit
        4.12
        test
    


    
        
        
            org.apache.tomcat.maven
            
            tomcat7-maven-plugin
            
                8001
                /
            
            
                
                    
                    package
                    
                        run
                    
                
            
        
    

注意pom此处设置了package后自动运行tomcat，并设置了端口号8001

mybatis配置文件

spring配置文件

web.xml

    
        springMVC
        org.springframework.web.servlet.DispatcherServlet
        
            contextConfigLocation
            classpath:spring/spring.xml
        
        1
        true
    
    
        springMVC
        /
    

实体类

//生成序列化
import java.io.Serializable;

public class Order implements Serializable {

    private String id;
    private Integer pid;
    private Integer userid;
}

import java.io.Serializable;


public class Product implements Serializable {

    private Integer id;
    private String product_name;
    private Integer stock;
    private Integer version;
}

Dao层

订单Dao

@Mapper
@Component
public interface OrderMapper {
    // 生成订单
    @Insert("insert into `order` (id,pid,userid) values (#{id},#{pid},#{userid})")
    int insert(Order order);
}

商品Dao

@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);
}

Service层

订单服务实现类

@Service
public class OrderServiceImpl implements OrderService {
    @Autowired
    ProductMapper productMapper;
    @Autowired
    OrderMapper orderMapper;
    @Override
    public void reduceStock(int id) throws Exception {
        // 1.获取库存
        Product product = productMapper.getProduct(id);
        // 模拟网络延迟
        Thread.sleep(1000);
        if(product.getStock() <= 0)
            throw new RuntimeException("已抢光！");
        // 2.减库存
        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("减库存失败，请重试！");
    }
}

Controller

@Controller
public class ProductAction {
    @Autowired
    private OrderService orderService;
    
    @GetMapping("/product/reduce")
    @ResponseBody
    public Object reduceStock(int id) throws Exception{
        orderService.reduceStock(id);
        return "ok";
    }
}

3. 启动测试有并发导致的错误

启动两次工程，端口号分别8001和8002,通过在pom中修改build的端口配置后，然后再次进行打包调用maven的package使用nginx做负载均衡

upstream sga{
    # 由于tomcat是在本机localhost启动的，所以此处的ip地址是本机的ip，可启动cmd用ipconfig指令查看
    server 192.168.204.1:8001;
    server 192.168.204.1:8002;
}

server {
    listen 80;
    server_name localhost;
    #charset koi8-r;
    #access_log logs/host.access.log main;
    location / {
        proxy_pass http://sga;
        root html;
        index index.html index.htm;
	}
}

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

下载地址：http://jmeter.apache.org/download_jmeter.cgi

注意此处写的Nginx的ip地址为服务器的地址

查看测试结果，10次请求全部成功查看数据库，stock库存变成 -5 （并发导致的数据结果错误）

4. apahce提供的zookeeper客户端

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

Curator：http://curator.apache.org/

    org.apache.curator
    curator-recipes
    4.2.0 

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

5. 在控制层中加入分布式锁的逻辑代码

@Controller
public class ProductAction {
    @Autowired
    private ProductService productService;
    private static String connectString =
        "192.168.204.141:2181,192.168.204.142:2181,192.168.204.143:2181";
    @GetMapping("/product/reduce")
    @ResponseBody
    public Object reduce( int id) throws Exception {
        // 重试策略 （1000毫秒试1次，最多试3次）
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        //1.创建curator工具对象
        Curatorframework client =
            CuratorframeworkFactory.newClient(connectString, retryPolicy);
        client.start();
        //2.根据工具对象创建“内部互斥锁”
        InterProcessMutex lock = new InterProcessMutex(client, "/product_"+id);
        try {
            //3.加锁
            lock.acquire();
            productService.reduceStock(id);
        }catch(Exception e){
            if(e instanceof RuntimeException){
                throw e;
            }
        }finally{
            //4.释放锁
            lock.release();
        }
        return "ok";
    }
}

正确的逻辑代码

@Controller
public class ProductAction {
    @Autowired
    private ProductService productService;
    private static String connectString =
        "192.168.204.141:2181,192.168.204.142:2181,192.168.204.143:2181";
    @GetMapping("/product/reduce")
    @ResponseBody
    public Object reduce( int id) throws Exception {
        // 重试策略 （1000毫秒试1次，最多试3次）
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        //1.创建curator工具对象
        Curatorframework client =
            CuratorframeworkFactory.newClient(connectString, retryPolicy);
        client.start();
        //2.根据工具对象创建“内部互斥锁”
        InterProcessMutex lock = new InterProcessMutex(client, "/product_"+id);
        try {
            //3.加锁
            lock.acquire();
            productService.reduceStock(id);
        }catch(Exception e){
            if(e instanceof RuntimeException){
                throw e;
            }
        }finally{
            //4.释放锁
            lock.release();
        }
        return "ok";
    }
}

再次测试，并发问题解决！