分布式锁从0到1落地实现01（mysql/redis/zk）

本文主要是介绍分布式锁从0到1落地实现01（mysql/redis/zk），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1 准备数据库表

CREATE TABLE `user` (
`id` bigint(20) NOT NULL COMMENT '主键ID',
`name` varchar(30) DEFAULT NULL COMMENT '姓名',
`age` int(11) DEFAULT NULL COMMENT '年龄',
`email` varchar(50) DEFAULT NULL COMMENT '邮箱',
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;INSERT INTO user (id, name, age, email) VALUES
(1, '杨幂', 18, 'yangmi@baomidou.com'),
(2, '刘亦菲', 20, 'yifei@baomidou.com'),
(3, '刘德华', 28, 'andy@baomidou.com'),
(4, '李嘉欣', 21, 'candy@baomidou.com'),
(5, '张国荣', 24, 'brother@baomidou.com');

– DestributeLock.stock definition

CREATE TABLE stock (
id bigint NOT NULL AUTO_INCREMENT,
product_code varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL,
ware_house varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL,
count int DEFAULT 0,
version int DEFAULT 0,
PRIMARY KEY (id),
KEY stock_product_code_IDX (product_code) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

2 官网

https://baomidou.com/pages/24112f/#%E6%A1%86%E6%9E%B6%E7%BB%93%E6%9E%84
https://baomidou.com/pages/223848/#keysequence
https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-data-redis/3.2.3

3 本地JVM锁失效的情况以及解决方案

   1:  //本地线程锁失效的案例 1 使用多例模式 并且指定代理模式为  CGLIB// @Scope(value = "prototype", proxyMode = ScopedProxyMode.TARGET_CLASS)2:  @Transactionalpublic synchronized void reduce() {........}或     @Transactionalpublic void reduce() {lock.lock();。。。。}由于事务注解 是是使用 AOP 来使用的，多线程情况下 在第一个线程还没提交事务之前 第二个先线程获取锁并且 在第一个线程之前执行完业务 并且提交完成 ，这时 第一个线程才 把数据提交 ，这样就会导致 两个线程修改之后的数据都是一样的 ，就会产生并发问题3: 在集群环境下，由于 请求负载到不同的 进程的服务了 ，这时jvm 的 事务也被隔离了 ，本地锁也只能管到自己了
===========================================================================4: 解决上述问题的方案一
使用数据库本身的锁，使用一个sql 在执行修改库存的时候 带上具体的查询条件public void reduce() {stockMapper.updateStock("1001",1);}@Update("update stock set count=#{count} where product_code=#{productCode} and count >= #{count}")int updateStock(@Param("productCode") String productCode, @Param("count") Integer count);缺点:这种方案没办法获取到锁修改前后的状态同一个商品有多个库存的时候没法判断修改哪一个所得范围是表锁5: 悲观锁   没加索引的时候锁范围还或者是全表，会有时所问题，加锁时枷锁的顺序要一致![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/74ce74a902554a67a0a57dd4945e72a7.png)@Transactionalpublic void reduce() {List<Stock> stocks = stockMapper.queryStockList("1001");Stock stock = stocks.get(0);if(stock != null && stock.getCount() > 0){stock.setCount(stock.getCount() -1);stockMapper.updateById(stock);}}@Select("select * from stock where product_code=#{productCode} for update")List<Stock> queryStockList(@Param("productCode") String productCode);添加索引之前的性能![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/f629813ae41641faaf8e7c99480b4158.png)
添加索引之后的性能
![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/cdba8ecd841f4d4ebac0be1fed031801.png)6：乐观锁  添加时间戳或者版本号 +  CAS 操作来实现 compare and swap 比较并交换
select * from stock where product_code ='10010' 
拿到version 字段
判断库存大于要购买的数量N
update stock set count = count -n ，version = 上一步查询出来的version + 1 and version= 上一步查询出来的version
如果影响的行为0 说明已经被别人改过了，需要循环或者递归重试@Transactionalpublic void reduce() {List<Stock> stocks = stockMapper.selectList(new QueryWrapper<Stock>().eq("product_code", "1001"));Stock stock = stocks.get(0);Integer version = stock.getVersion();if (stock != null && stock.getCount() > 0) {stock.setCount(stock.getCount() - 1);stock.setVersion(version + 1);if (stockMapper.update(new UpdateWrapper<Stock>().eq("id", stock.getId()).eq("version", stock.getVersion())) == 0) {reduce();}}}
![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/91ad0c4c8e124209be59b1321b6fe428.png)

4 redis 版本的锁

1： 添加依赖<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><version>3.2.3</version></dependency>
2： 修改配置文件
spring.data.redis.host=192.168.187.128
spring.data.redis.port=6379
spring.data.redis.database=0
spring.cache.type=REDIS
spring.data.redis.timeout=10000
spring.data.redis.lettuce.pool.max-active=100
spring.data.redis.lettuce.pool.max-wait=-1
spring.data.redis.lettuce.pool.max-idle=8
spring.data.redis.lettuce.pool.min-idle=0
spring.data.redis.lettuce.pool.time-between-eviction-runs=10s编写一个测试用例测试是否redis正确连接此时可能会出现 连不上redis ，需要修改配置文件
注释 掉 bind 127.0.0.1 这个项
可以连接上之后 关闭服务器再打开会发现我们set 的 值消失了 这是因为我们没有开启持久化
可以去开启 rdb 或者 aof 我这里开启了 aof 
在配置文件中将 appendonly 的 no 改为 yes 
重启服务  ./redis-server redis.conf
现在去运行代码可以看到 正常运行结果了，环境搭建完成
![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/ba77bd876acf40998d981548d6a2669c.png)
![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/5ee09450d4794419b8949d52ff7daefb.png)3: 使用 redis 自带的乐观锁来实现锁结局超卖问题
可以保证线程安全问题但是，但是并发量非常低，性能不能保证,不推荐使用public void reduce() {this.redisTemplate.execute(new SessionCallback<Object>() {@Overridepublic  Object execute(RedisOperations operations) throws DataAccessException {operations.watch("stock"); //监控节点String stock = operations.opsForValue().get("stock")==null?"": operations.opsForValue().get("stock").toString();//判断库存是否充足if(StringUtils.isNoneBlank(stock) && Integer.parseInt(stock) != 0){Integer st = Integer.parseInt(stock);if(st > 0){operations.multi(); //开启事务//扣减库存operations.opsForValue().set("stock",String.valueOf(--st));//执行事务List exec = operations.exec();if(exec == null || exec.size()==0){//如果返回为空则表示扣减库存失败需要重试try {Thread.sleep(20);} catch (InterruptedException e) {throw new RuntimeException(e);}reduce();}return exec;}}return null;}});}

这篇关于分布式锁从0到1落地实现01（mysql/redis/zk）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---