zookeeper案例-分布式锁-商品秒杀原理解析

2023-12-30 14:18:58

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

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

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

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

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

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

1. 实现步骤

1.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	-- 用户编号
)

1.2 搭建工程

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

具体代码见github

1.3 启动测试

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

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

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

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

1.4 apahce提供的zookeeper客户端

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

Curator：h ttp://curator.apache.org/

<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>4.2.0</version> <!-- 网友投票最牛逼版本 -->
</dependency>

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

1.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";
}
}

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

节选自拉钩教育JAVA系列课程

码农公寓