分布式锁(rediszk)

Posted 2021-06-30 xue_yun_xiang

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了分布式锁(rediszk)相关的知识，希望对你有一定的参考价值。

一、概述

什么是锁？

synchronized ,lock 都是锁

锁的作用

保证当前线程访问该方法或者代码块时，方法内、代码块中的资源（共享变量）同一时刻只有一个线程在访问，线程安全

synchronized 、lock的特性

1.保证synchronized ,lock 代码都是同一时刻单线程运行
2.同时能够保证代码块内的 共享变量具有可见性，保证其他线程能够立即看到代码块内的共享变量的值

二、volatile

内存语义两个

可见性：当一个线程修改了某一个变量值，这个值会被其他线程立即可见
禁止指令重排：禁止jvm 优化代码的执行顺序

int a = 1;
int b = 1;
inc = a +b;
jvm 先运行 int a = 1; 或者 int b = 1; 对于代码执行结果没有影响，但是jvm在给你保证单线程符合你的代码逻辑情况下，进行优化有可能先执行 int b = 1; 在执行 int a = 1;
https://www.jianshu.com/p/a9ecac4bd753
如果是单例模式
public class DoubleCheckedLocking { //1
private static Instance instance; //2
public static Instance getInstance(){ //3
if(instance ==null) { //4:第一次检查
synchronized (DoubleCheckedLocking.class) { //5：加锁
if (instance == null) //6：第二次检查
instance = new Instance(); //7：问题的根源处在这里
} //8
} //9
return instance; //10
} //11
}
instance = new Instance(); 这一行代码在jvm 底层有多行执行
memory=allocate(); //1:分配对象的内存空间
ctorInstance(memory); //2:初始化对象
instance = memory; //3:设置instance指向刚分配的内存地址
jvm 优化以后，有没有有可能
memory=allocate(); //1:分配对象的内存空间
instance = memory; //3:设置instance指向刚分配的内存地址
此时没有初始化对象，如果被其他线程看到，会造成获取一个没有初始化的对象，造成异常
如何解决？将 instance 使用 volatile 修饰，禁止执行重排禁止创建对象的三步优化，

在这里插入图片描述

三、实现

1、创建工程，引入依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.4.RELEASE</version>
</parent>


<dependencies>

    <!--springBoot 相关  -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
</dependencies>

2、启动类

3、商品秒杀

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * 商品 秒杀相关
 */

@RestController
public class GoodsController {

    @Autowired
    private StringRedisTemplate redisTemplate;


    /**
     * 初始化商品订单 库存
     * @param goods
     * @return
     */
    @RequestMapping("/init")
    public String init(String goods){

        // 初始化 stack 为 1000  订单 0

        redisTemplate.opsForValue().set("stack_"+goods, 1000+"");
        redisTemplate.opsForValue().set("order_"+goods, 0+"");



        return "当前商品 " +goods +" 库存："+ redisTemplate.opsForValue().get("stack_"+goods)+"--订单："
                + redisTemplate.opsForValue().get("order_"+goods);
    }

    // synchronized 加锁 保证 当前应用 单线程 执行 kill ,保证线程安全
    // synchronized 只能保证单个应用的线程安全，如果 对应的代码块的资源被多个引用，
    // 则synchronized 不能起到线程安全的作用，此时需要一个全局的，夸应用的锁--》分布式锁（zk(临时有序节点) ,redis(setnx)）
    @RequestMapping("/kill")
    public synchronized String kill(String goods){

        // 1.获取当前的库存  并判断 数量
        Integer currentStack = Integer.valueOf( redisTemplate.opsForValue().get("stack_"+goods));

        if (currentStack <=0 ){ // 没有库存

            return  "抱歉，已售罄" + goods;
        }


        // 2.削减库存
        redisTemplate.opsForValue().set("stack_"+goods, (currentStack -1)+"");

        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 3.订单增加 1
        Integer order = Integer.valueOf( redisTemplate.opsForValue().get("order_"+goods));
        redisTemplate.opsForValue().set("order_"+goods, (order +1 )+"");


        return "已抢到到该商品" + goods;
    }
    /**
     * 获取 当前商品 对应的 库存 和 订单 数量
     * @param goods
     * @return
     */
    @RequestMapping("/getState")
    public String getState(String goods){

        return "当前商品 " +goods +" 库存："+ redisTemplate.opsForValue().get("stack_"+goods)+"--订单："
                + redisTemplate.opsForValue().get("order_"+goods);
    }
}

4、始化商品库存

http://localhost:8888/init?goods=mate50

5、使用ab 工具启动两个应用 8888 8889 分别发起秒杀测试

在这里插入图片描述

ab -n 1000 -c 2 http://localhost:8888/kill?goods=mate50
ab -n 请求数 -c 并发数访问的路径

6、带秒杀完毕出现结果不一致，数据不安全

在这里插入图片描述

四、使用 redis 实现分布式锁

1、创建RedisLock 实现类

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
public class RedisLock {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;


    /**
     * 获取锁的方法
     */
    public boolean lock(String key,String value,int time){

       // 如果能够设置key 成功，key 不存在  -----》意义就是当前请求获取了 锁
       //       设置 key  失败，key 存在  -------》意义 已经有其他请求持有 锁
       return stringRedisTemplate.opsForValue().setIfAbsent(key, value, time, TimeUnit.SECONDS );
    }


    /**
     * 删除 key  代表 释放锁
     * @param key
     * @return
     */
    public boolean unlock(String key){

        return stringRedisTemplate.delete(key);
    }
}

2、使用

 @RequestMapping("/redisKill")
    public  String redisKill(String goods){

        if (redisLock.lock(goods, "1",  3)){ // 成功 获取锁

            // 1.获取当前的库存  并判断 数量
            Integer currentStack = Integer.valueOf( redisTemplate.opsForValue().get("stack_"+goods));

            if (currentStack <=0 ){ // 没有库存

                return  "抱歉，已售罄" + goods;
            }


            // 2.削减库存
            redisTemplate.opsForValue().set("stack_"+goods, (currentStack -1)+"");

            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            // 3.订单增加 1
            Integer order = Integer.valueOf( redisTemplate.opsForValue().get("order_"+goods));
            redisTemplate.opsForValue().set("order_"+goods, (order +1 )+"");

            //释放锁
            redisLock.unlock(goods);

            return "已抢到到该商品" + goods;
        }else {

            return "抱歉抢购该商品" + goods +"失败";
        }


    }

五、使用zk 实现分布式锁

在这里插入图片描述

但是，使用zk 本身的监控功能去实现分布式锁，非常繁琐，所以使用第三方封装的锁
InterProcessMutex 提供互斥锁来自 curator 框架

1、引入 curator 依赖

<!--  zk 依赖-->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.6.0</version>
        </dependency>

        <!--zk  封装  -->
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.1</version>
        </dependency>

2、将CuratorFramework 加入到容器

@Configuration
public class ZkConfig {



    /**
     * CuratorFramework 封装了 zkClient
     * @return
     */
    @Bean // 让
    public  CuratorFramework createCuratorFramework(){

        // 客户端重试策略 如果连不上zk server 会发起重试
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,2);

        CuratorFramework curatorFramework =
                CuratorFrameworkFactory.newClient("192.168.12.130:2181,192.168.12.130:2182,192.168.12.130:2183", retryPolicy);

        // 发起连接
        curatorFramework.start();

        return  curatorFramework;
    }
}

3、使用zk 作为分布式锁

/**
 *  使用zk  完成分布式锁
 *
 *
 * @param goods
 * @return
 * @throws Exception
 */
@RequestMapping("/zkKill")
public  String zkKill(String goods) throws Exception {

    // interProcessMutex 就是创建对 zk 分布式锁的封装
    InterProcessMutex interProcessMutex = new InterProcessMutex(curatorFramework,"/"+goods);
    // interProcessMutex.acquire 获取锁 ，并且最多持有3 s
    //  true 代表 持有锁
    //  false  代表持有锁失败
    if (interProcessMutex.acquire(3, TimeUnit.SECONDS)){ // 成功 获取锁

        // 1.获取当前的库存  并判断 数量
        Integer currentStack = Integer.valueOf( redisTemplate.opsForValue().get("stack_"+goods));

        if (currentStack <=0 ){ // 没有库存

            return  "抱歉，已售罄" + goods;
        }


        // 2.削减库存
        redisTemplate.opsForValue().set("stack_"+goods, (currentStack -1)+"");

        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 3.订单增加 1
        Integer order = Integer.valueOf( redisTemplate.opsForValue().get("order_"+goods));
        redisTemplate.opsForValue().set("order_"+goods, (order +1 )+"");

        //释放锁  释放zk 分布式锁 其实就是删除自己创建的临时节点
        interProcessMutex.release();

        return "已抢到到该商品" + goods;
    }else {

        return "抱歉抢购该商品" + goods +"失败";
    }


}

4、zk vs redis 分布式锁

实际工作首选redis,因为 redis 基于内存并发量很高，不易成为性能瓶颈，zk 的存储基于文件（磁盘），并发写的能力很差，另外zk 创建节点是不仅master要写入，slave 也要写入，从而更加耗时，所以zk 只适合并发量很低的场合。

以上是关于分布式锁(rediszk)的主要内容，如果未能解决你的问题，请参考以下文章

分布式锁(rediszk)

分布式锁（数据库RedisZK）拍了拍你

Redis实现分布式锁（设计模式应用实战）

分布式锁三种解决方案

分布式锁(rediszk)

一、概述

什么是锁？

锁的作用

synchronized 、lock的特性

二、volatile

三、实现

1、创建工程，引入依赖

2、启动类

3、商品秒杀

4、始化商品库存

5、使用ab 工具 启动 两个应用 8888 8889 分别发起秒杀测试

6、带秒杀完毕出现结果不一致，数据 不安全

四、使用 redis 实现分布式锁

1、创建RedisLock 实现类

2、使用

五、使用zk 实现分布式锁

1、引入 curator 依赖

2、将CuratorFramework 加入到容器

3、使用zk 作为分布式锁

4、zk vs redis 分布式锁

5、使用ab 工具启动两个应用 8888 8889 分别发起秒杀测试

6、带秒杀完毕出现结果不一致，数据不安全