Zookeeper_随笔_内存溢出

Zookeeper

一、Zookeeper概述
- 1.概述
- 2.工作机制
- 3.特点
- 4.数据结构
- 5.应用场景
- - ①统一命名服务
  - ②统一配置管理
  - ③服务器节点动态上下线
  - ④软负载均衡
- 6.下载地址
二、Zookeeper本地模式安装
- 1.本地模式安装
- - ①安装前准备
  - ②配置修改
  - ③ *** 作Zookeeper
- 2.配置参数解读
三、Zookeeper内部原理
- 1.选举机制（面试重点）
- 2.节点类型
- 3.监听器原理（面试重点）
- 4. 写数据流程
四、Zookeeper实战
- 1.分布式安装部署
- - ①安装zookeeper
  - ②配置服务器编号
  - ③配置zoo.cfg文件
  - ④配置其余两台服务器
  - ⑤集群 *** 作
- 2.客户端命令行 *** 作
- 3.API应用
- - ①IDEA环境搭建
  - ②创建ZooKeeper客户端
  - ③创建节点
  - ④查询节点的值
  - ⑤修改节点的值
  - ⑥删除节点
  - ⑦获取子节点
  - ⑧监听子节点的变化
  - ⑨判断Znode是否存在
- 4. 案例-模拟美团商家上下线
- - ①需求
  - ②商家服务类
  - ③客户类
- 5.案例-分布式锁-商品秒杀
- - ①初始化数据库
  - ②搭建工程
  - ③启动测试
  - ④apahce提供的zookeeper客户端
  - ⑤在控制层中加入分布式锁的逻辑代码
  - ⑥再次测试，并发问题解决！

一、Zookeeper概述 1.概述

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

2.工作机制

Zookeeper从设计模式角度来理解：是一个基于观察者模式（一个人干活，有人盯着他）设计的分布式服务管理框架
它负责存储和管理大家都关心的数据
○ 然后接受观察者的注册，一旦这些数据的发生变化
○ Zookeeper就将负责通知已经注册的那些观察者做出相应的反应
○ 从而实现集群中类似Master/Slave管理模式
Zookeeper = 文件系统 + 通知机制

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

3.特点

分布式和集群的区别
○ 无论分布式和集群，都是很多人在做事情。具体区别如下：
○ 例如：我有一个饭店，越来越火爆，我得多招聘一些工作人员
　　◆ 分布式：招聘1个厨师，1个服务员，1个前台，三个人负责的工作不一样，但是最终目的都是为饭店工作
　　◆ 集群：招聘3个服务员，3个人的工作一样

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

4.数据结构

ZooKeeper数据模型的结构与linux文件系统很类似，整体上可以看作是一棵树，每个节点称做一个ZNode（ZookeeperNode）。
每一个ZNode默认能够存储1MB的数据（元数据），每个ZNode的路径都是唯一的
○ 元数据（metadata），又称中介数据、中继数据，为描述数据的数据（data about data），主要是描述数据属性（property）的信息，用来支持如指示存储位置、历史数据、资源查找、文件记录等功能

5.应用场景

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

①统一命名服务

在分布式环境下，通常需要对应用或服务进行统一的命名，便于识别
例如：服务器的IP地址不容易记，但域名相比之下却是很容易记住

②统一配置管理

分布式环境下，配置文件做同步是必经之路
1000台服务器，如果配置文件作出修改，那一台一台的修改，运维人员肯定会疯，如何做到修改一处就快速同步到每台服务器上
将配置管理交给Zookeeper
◇ 将配置信息写入到Zookeeper的某个节点上
◇ 每个客户端都监听这个节点
◇ 一旦节点中的数据文件被修改，Zookeeper这个话匣子就会通知每台客户端服务器

③服务器节点动态上下线

客户端能实时获取服务器上下线的变化
在美团APP上实时可以看到商家是否正在营业或打样

④软负载均衡

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

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包

二、Zookeeper本地模式安装 1.本地模式安装 ①安装前准备

1.安装jdk
2.拷贝apache-zookeeper-3.6.0-bin.tar.gz到opt目录
3.解压安装包

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

4.重命名

[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

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

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

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

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

③ *** 作Zookeeper

1.启动Zookeeper

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

2.查看进程是否启动

[root@localhost bin]# jps

QuorumPeerMain：是zookeeper集群的启动入口类，是用来加载配置启动QuorumPeer线程的

3.查看状态

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

4.启动客户端

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

5.退出客户端

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

2.配置参数解读

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

tickTime =2000：通信心跳数，Zookeeper服务器与客户端心跳时间，单位毫秒
◆ Zookeeper使用的基本时间，服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个tickTime时间就会发送一个心跳，时间单位为毫秒。
initLimit =10：LF初始通信时限
◆ 集群中的Follower跟随者服务器与Leader领导者服务器之间，启动时能容忍的最多心跳数
◆ 10*2000（10个心跳时间）如果领导和跟随者没有发出心跳通信，就视为失效的连接，领导和跟随者彻底断开
syncLimit =5：LF同步通信时限
◆ 集群启动后，Leader与Follower之间的最大响应时间单位，假如响应超过syncLimit * tickTime->10秒，Leader就认为Follwer已经死掉，会将Follwer从服务器列表中删除
dataDir：数据文件目录+数据持久化路径
◆ 主要用于保存Zookeeper中的数据。
dataLogDir：日志文件目录
clientPort =2181：客户端连接端口
◆ 监听客户端连接的端口。

三、Zookeeper内部原理 1.选举机制（面试重点）

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

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

2.节点类型

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

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

3.监听器原理（面试重点）

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

4. 写数据流程

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

四、Zookeeper实战 1.分布式安装部署

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

①安装zookeeper

请参考本文 [ 二、Zookeeper本地模式安装 ]

②配置服务器编号

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

[root@localhost zkData]# vim myid

在文件中添加与server对应的编号：1
其余两台服务器分别对应2和3

③配置zoo.cfg文件

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

#######################cluster##########################
server.1=192.168.115.81:2888:3888
server.2=192.168.115.82:2888:3888
server.3=192.168.115.83:2888:3888

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

④配置其余两台服务器

在虚拟机数据目录vms下，创建zk02
将本台服务器数据目录下的.vmx文件和所有的.vmdk文件分别拷贝zk02下
虚拟机->文件->打开（选择zk02下的.vmx文件）
开启此虚拟机，d出对话框，选择“我已复制该虚拟机”
进入系统后，修改linux中的ip，修改/opt/zookeeper/zkData/myid中的数值为2

第三台服务器zk03，重复上面的步骤

⑤集群 *** 作

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

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

2.启动第1台

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

3.查看状态

[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.

注意：因为没有超过半数以上的服务器，所以集群失败（防火墙没有关闭也会导致失败）
4.当启动第2台服务器时
查看第1台的状态：Mode: follower
查看第2台的状态：Mode: leader

2.客户端命令行 *** 作

启动客户端

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

显示所有 *** 作命令

help

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

ls /

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

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 "cold"

获得节点的值

get /japan/Tokyo

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

create -e /uk
ls /
quit
ls /

创建带序号的节点

在俄罗斯ru下，创建3个city

create -s /ru/city # 执行三次
ls /ru
[city0000000000, city0000000001, city0000000002]

如果原来没有序号节点，序号从0开始递增。
如果原节点下已有2个节点，则再排序时从2开始，以此类推

修改节点数据值

set /japan/Tokyo "too cold"

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

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

addWatch /usa

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

set /usa "baideng"

Server3会立刻响应
WatchedEvent state:SyncConnected type:NodeDataChanged path:/usa
如果在Server1的/usa下面创建子节点NewYork

create /usa/NewYork

Server3会立刻响应

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

删除节点

delete /usa/NewYork

递归删除节点（非空节点，节点下有子节点）

deleteall /ru

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

3.API应用 ①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

②创建ZooKeeper客户端

	//zookeeper集群的ip和端口
    private String connectionString = "192.168.117.128:2181,192.168.117.129:2181,192.168.117.130:2181";

    //session超时的时间
    //时间不宜设置太小 因为连接zookeeper和加载集群环境会因为性能原因延迟略高
    //如果时间太少，还没有创建好客户端，就开始 *** 作节点，会报错的
    private int sessionTimeout = 60*1000;

    //zookeeper客户端对象
    private ZooKeeper zkClient;

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

③创建节点

一个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 Exception {
        zkClient = new ZooKeeper(connectionString, sessionTimeout, new Watcher() {
            public void process(WatchedEvent watchedEvent) {
                System.out.println("得到监听反馈，再进行业务处理！");
                System.out.println(watchedEvent.getType());
            }
        });
    }

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

④查询节点的值

	//获取节点上的值
    @Test
    public void getNodeData() throws Exception{
        byte[] bytes = zkClient.getData("/zk", false, new Stat()); // 路径不存在时会报错
        String str = new String(bytes);
        System.out.println("/zk节点的数据：" + str);
    }

⑤修改节点的值

	//修改节点上的值
    @Test
    public void update() throws Exception{
        Stat stat = zkClient.setData("/zk", "AMEameame".getBytes(), 0); //先查看节点详情，获得dataVersion = 0
        System.out.println(stat);
    }

⑥删除节点

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

⑦获取子节点

	//获取子节点
    @Test
    public void getChildren() throws Exception {
        List children = zkClient.getChildren("/china",false); // false:不监听
        for (String child : children) {
            System.out.println(child);
        }
    }

⑧监听子节点的变化

	//监听根节点下面的变化
    @Test
    public void watchChildren() throws Exception {
        List children = zkClient.getChildren("/", true); // true：注册监听
        for (String child : children) {
            System.out.println(child);
        }
        // 让线程不停止，等待监听的响应
        //Thread.sleep(1000000000);
        System.in.read();
    }

⑨判断Znode是否存在

	//判断Znode是否存在
    @Test
    public void exist() throws Exception {
        Stat stat = zkClient.exists("/zk", false);
        System.out.println(stat == null ? "不存在" : "存在");
    }

4. 案例-模拟美团商家上下线 ①需求

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

②商家服务类

public class ShopServer {
    private String connectionString = "192.168.117.128:2181,192.168.117.129:2181,192.168.117.130:2181";
    private int sessionTimeout = 60 * 1000;
    private ZooKeeper zkClient;
    public static void main(String[] args) throws Exception {
        ShopServer shop = new ShopServer();
        //1.连接zookeeper集群
        shop.getConnection();
        //2.将服务节点注册到zookeeper
        shop.register(args[0]);
        //3.业务逻辑处理
        shop.business(args[0]);
    }
    //创建客户端，连接到zookeeper
    public void getConnection() throws Exception{
        zkClient = new ZooKeeper(connectionString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {

            }
        });
    }
    //注册到zookeeper
    public void register(String shopName) throws Exception{
        //EPHEMERAL_SEQUENTIAL临时有序的节点
        String s = zkClient.create("/meituan/shop", shopName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println("[" + shopName + "]" + "开始营业了！" + s);
    }
    //做买卖
    public void business(String shopName) throws Exception{
        System.out.println("[" + shopName + "]正在营业中!");
        System.in.read();
    }
}

③客户类

public class Customers {
    private String connectionString = "192.168.117.128:2181,192.168.117.129:2181,192.168.117.130:2181";
    private int sessionTimeout = 60 * 1000;
    private ZooKeeper zkClient;
    public static void main(String[] args) throws Exception {
        Customers client = new Customers();
        //1.获取zookeeper连接
        client.getConnection();
        //2.获取meituan下的所有子节点
        client.getShopList();
        //3.业务逻辑处理
        client.business();
    }
    //创建客户端，连接到zookeeper
    public void getConnection() throws Exception{
        zkClient = new ZooKeeper(connectionString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                //再次获取商家列表
                try {
                    getShopList();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
    //再次获取商家列表
    private void getShopList() throws Exception {
        //1.获取服务器子节点信息，并对父节点进行监听
        List shops = zkClient.getChildren("/meituan", true);
        //2.声明存储服务器信息的集合
        ArrayList shopList = new ArrayList<>();
        for (String shop : shops) {
            byte[] bytes = zkClient.getData("/meituan/" + shop, false, new Stat());
            shopList.add(new String(bytes));
        }
        System.out.println("目前正在营业的商家 " + shopList);
    }
    private void business() throws Exception {
        System.out.println("用户正在浏览商家。。。");
        System.in.read();
    }
}

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

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

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

delete /meituan/baozi

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

5.案例-分布式锁-商品秒杀

锁：我们在多线程中接触过，作用就是让当前的资源不会被其他线程访问！
　　○ 我的日记本，不可以被别人看到。所以要锁在保险柜中
　　○ 当我打开锁，将日记本拿走了，别人才能使用这个保险柜
在zookeeper中使用传统的锁引发的 “羊群效应” ：1000个人创建节点，只有一个人能成功，999人需要等待！
羊群是一种很散乱的组织，平时在一起也是盲目地左冲右撞，但一旦有一只头羊动起来，其他的羊也会不假思索地一哄而上，全然不顾旁边可能有的狼和不远处更好的草。羊群效应就是比喻人都有一种从众心理，从众心理很容易导致盲从，而盲从往往会陷入骗局或遭到失败。
避免“羊群效应”，zookeeper采用分布式锁

所有请求进来，在/lock下创建临时顺序节点，放心，zookeeper会帮你编号排序
判断自己是不是/lock下最小的节点
a.是，获得锁（创建节点）
b.否，对前面小我一级的节点进行监听
获得锁请求，处理完业务逻辑，释放锁（删除节点），后一个节点得到通知（比你年轻的死了，你成为最嫩的了）
重复步骤2

实现步骤

①初始化数据库

创建数据库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 -- 用户编号
)

②搭建工程

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



    
    4.0.0
    com.zk
    zk_product
    1.0-SNAPSHOT
    war
    
        5.2.7.RELEASE
        
        UTF-8
        
        UTF-8
        11
        11
        11
    
    
        
        
            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.curator
            curator-recipes
            4.2.0 
        
    
    
        
            
            
                org.apache.tomcat.maven
                
                tomcat7-maven-plugin
                
                    8001
                    /
                
                
                    
                        
                        package
                        
                            run




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

public class Order implements Serializable {
    private String id;
    private int pid;
    private int userid;
    getter/setter 
}

public class Product implements Serializable {
    private int id;
    private String product_name;
    private int stock;
    private int version;
    getter/setter 
}

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

@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
public class ProductServiceImpl implements ProductService {
    @Autowired
    ProductMapper productMapper;
    @Autowired
    OrderMapper orderMapper;
    public void reduceStock(int id) throws Exception {
        // 1.获取库存(根据商品id查询商品)
        Product product = productMapper.getProduct(id);
        // 模拟网络延迟
        //Thread.sleep(1000);
        if(product.getStock() <= 0){
            throw new RuntimeException("已抢光！");
        }
        // 2.减库存
        int i = productMapper.reduceStock(id);
        if(i == 1){
            //3.生成订单
            Order order = new Order();
            //UUID工具类生成订单号
            order.setId(UUID.randomUUID().toString());
            order.setPid(id);
            order.setUserid(101);
            orderMapper.insert(order);
        }else{
            throw new RuntimeException("减库存失败，请重试！");
        }
    }
}

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

③启动测试

启动两次工程，端口号分别8001和8002
使用nginx做负载均衡

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

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

下载地址：http://jmeter.apache.org/download_jmeter.cgi
点击此处进行下载

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

④apahce提供的zookeeper客户端

基于zookeeper原生态的客户端类实现分布式是非常麻烦的，我们使用apahce提供了一个zookeeper客户端来实现
Curator：http://curator.apache.org/
点击此处下载


	org.apache.curator
	curator-recipes
	4.2.0

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

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

@Controller
public class ProductAction {
    @Autowired
    private ProductService productService;
    private static String connectString = "192.168.117.128:2181,192.168.117.129:2181,192.168.117.130:2181";
    @GetMapping("/product/reduce")
    @ResponseBody
    public Object reduceStock(int id) throws Exception{
        //1.创建curator工具对象
        // 重试策略 （1000毫秒试1次，最多试3次）
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        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";
    }
}

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

欢迎分享，转载请注明来源：内存溢出

原文地址: https://outofmemory.cn/zaji/5677453.html

Zookeeper

发表评论

评论列表（0条）