分布式锁从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');



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

– 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
1
2
3

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

   1:  //本地线程锁失效的案例 1 使用多例模式 并且指定代理模式为  CGLIB
   // @Scope(value = "prototype", proxyMode = ScopedProxyMode.TARGET_CLASS)
   2:  @Transactional
    public synchronized void reduce() {........}
        或     @Transactional
    public 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)

     @Transactional
    public 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 说明已经被别人改过了，需要循环或者递归重试
    @Transactional
    public 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)
 
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

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>() {
            @Override
            public  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;
            }
        });
    }
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