分布式锁的三种实现方式_分布式锁的lock实现方式有哪些

目录

 

一.分布式锁概念

二.Mysql实现分布式锁

1.原理

2.创建商品表tb_good

3.创建订单表tb_order

3.创建tb_lock表

4.加锁逻辑

5.使用锁

三.Redis分布式锁

1.原理

2.死锁问题

3.lua脚本实现加锁和释放锁

(1)获取锁

（2）在程序中编写一个Configuretion

（3）lua使用方式

4.看门狗

5.Redisson框架使用

 (1)引入依赖

(2)配置redis

6.红锁

(1)原理

（2）引入依赖

(3)配置redis

(3)使用redlock

7.redis分布式锁的问题

四.Zookeper分布式锁

1.原理

(1)获取锁

(2)释放锁

2.Zookeepr结合mysql乐观锁

3.代码实现

（1）引入依赖

（3）创建CuratorFrameWork

(4)使用

---

 

一.分布式锁概念

  在很多问题情境中，如多个用户抢购同一个商品，外卖员抢同一个用户的订单等，常会出现超卖问题，即每个线程都创建订单，导致超卖，为了解决超卖问题，需要对下单业务加锁

1.在单体项目中，可以使用JVM锁（如:synchronized,lock）可以实现

2.在分布式项目中，常用mysql,redis,zookeeper来实现分布式锁

二.Mysql实现分布式锁

1.原理

  根据主键和索引的唯一性，可以作为锁的key，当执行业务之前，向表中插入一条数据，当其他线程尝试插入数据时，发现已经存在，则插入失败，也就是获取锁失败，这样就可以作为分布式锁使用

 下面以抢购商品来演示分布式锁的使用

2.创建商品表tb_good

create table 'tb_good'(
  'good_id' int(16) not null,
  'num' int(8) default null,
  'update_time' timestamp null default current_timestamp on update current_timestamp,
   primary key ('goo_id')
 
) engine=InnoDB default charset=utf8

在数据库中插入一条商品信息：

insert into 'tb_good' values('1','2','2023-09-12 12:23:12');

3.创建订单表tb_order

 

create table 'tb_order'(
  'order_id' int(8) not null auto_increment,
	'order_status' int(8) default null,
	'order_description' varchar(128) character set utf8 collate
	utf8_general_ci default null,
	'user_id' int(8) default null,
	'update_time' TIMESTAMP null DEFAULT CURRENT_TIMESTAMP on update CURRENT_TIMESTAMP,
	PRIMARY KEY ('order_id')
)ENGINE=INNODB AUTO_INCREMENT=434 default charset=utf8;

先扣减商品中的库存，然后在订单表中新增一条订单记录

3.创建tb_lock表

 

create table tb_lock(
  'good_id' int(16) not null,
  'lock_start_time' datetime default null on update CURRENT_TIMESTAMP,
	'lock_end_time' datetime default null on update CURRENT_TIMESTAMP,
	primary key ('good_id')
)ENGINE=INNODB  default charset=utf8;

当抢购商品时，向表中插入一条记录，下完订单后删除记录

4.加锁逻辑

public class MysqlLock implements Lock {
 
    @Autowired
    private OrderDao mapper;
    private ThreadLocal<OrderDao> orderDaoThreadLocal;
    
    /**
     * 获取锁
     */
    @Override
    public void lock() {
        if(tryLock()){
            System.out.println("尝试加锁");
            return;
        }
        //休眠
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        //递归再次调用
        lock();
    }
 
    @Override
    public void lockInterruptibly() throws InterruptedException {
        
    }
    /**
     * 非阻塞式锁，成功就成功，失败就失败，直接返回
     * @return
     */
    @Override
    public boolean tryLock() {
        OrderDao orderDao = orderDaoThreadLocal.get();
        mapper.insertLock(orderDao);
        System.out.println("加锁对象:"+orderDaoThreadLocal.get());
        return true;
 
    }
 
    /**
     * 释放锁
     */
    @Override
    public void unlock() {
       mapper.deleteLock(orderDaoThreadLocal.get().getOrderId());
        System.out.println("解锁对象:"+orderDaoThreadLocal.get());
        orderDaoThreadLocal.remove();
    }
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }
    
    @Override
    public Condition newCondition() {
        return null;
    }
}

5.使用锁

在需要使用的地方如下操作：

//加锁
lock.lock();
//释放锁
lock.unlock();

三.Redis分布式锁

1.原理

 利用redis的互斥操作

set key value

当且仅当key不存在时，设置value

释放锁

del key

2.死锁问题

当使用setnx加锁后，锁将永远存在redis中，其他线程就无法继续获取锁，，若在执行完senx操作后，程序宕机了，无法删除Key,那么就会造成死锁，这时就得给key设置过期时间，需要让setnx和设置过期时间是一个原子操作，

set key value nx ex 10

3.lua脚本实现加锁和释放锁

(1)获取锁

--- 获取key
local key = KEYS[1]
--- 获取value
local value =KEYS[2]
---获取一个参数
local expire = ARGV[1]
--- 如果redis找不到这个key就去插入
if redis.call("get",key) == false then
    ---如果插入成功，就去设置过期时间
    if redis.call("set",key,value) then
        --- lua脚本接受到参数都会转化为String,所有要转化为数字进行比较
        if tonumber(expire) > 0 then
            ---设置过期时间
            redis.call("expire",key,expire)
        end
        return true
    end
    return false
else
    return false
end 

（2）在程序中编写一个Configuretion

@Configuration
public class LuaConfiguration {
    
    @Bean(name="set")
    public DefaultRedisScript<Boolean> redisScript(){
        
        DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new 
                ClassPathResource("lock-set.lua")));
        
        redisScript.setResultType(Boolean.class);
        return redisScript;
    }
    
}

使用时直接注入即可

删除锁逻辑lock-del.lua

if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end 

（3）lua使用方式

List<String> keys = Arrays.asList("testLua","hello lua");
Boolean execute = stringRedisTemplate.execute(redisScript,keys,"100");
return null;

这里的"testLua"和"hello lua" 分别对应上面的KEYS[1]和KEYS[2]

4.看门狗

 当设置的key有效期为10s,结果程序执行了15S，那么在最后5s，key已经过期了，如果有另外一个线程过来加锁，还是能加锁成功的，这样就有两个线程拿到了锁，依然会引起超卖问题

  还有一种情况，当第二个线程执行业务时，第一个线程开始执行释放锁，把第二个线程的锁释放了，这样以此类推，线程二会释放线程三的锁，三会释放四的锁，这样会导致程序继续错乱

如果，在锁快要过期的时候，业务还没执行完，这时有人给锁继续延长过期时间，那么就能保证业务执行完再释放锁，这种机制是看门狗WatchDog

遗留问题: 线程一删除线程二的锁，只需要给key设置一个独一无二的value值，这样在删除锁之前做一次判断，如果是自己的锁，就释放，如果不是自己的锁，不释放。

5.Redisson框架使用

 (1)引入依赖

     <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.13.6</version>
        </dependency>

(2)配置redis

public class RedisConfig {
    @Autowired
    private  RedisSentinelProperties properties;
 
   @Bean
    public RedissonClient redissonClient(){
       Config config = new Config();
       config.useSingleServer().setAddress("localhost:6379").setDatabase(0);
       return Redisson.create(config);
   }
 
 
 
}

在代码中注入RedissionClient

 @Autowired
    private RedissonClient redissonClient;
    
    public String fightOrder(int goodId,int userId){
        
        //生成key
        String  key = "goodsId_" + goodId+"";
 
        RLock rLock = redissonClient.getLock(key);
        
        try{
            //设置过期时间为30
            rLock.lock();
            
           boolean b = seckillOrderService.add(goodId,userId);
           if(b){
               System.out.println("用户:"+userId+"抢单成功！");
           }else {
               System.out.println("用户:"+userId+"抢单失败!");
           }
        }finally {
            rLock.unlock();
            
        }
        return null;
        
    }

对于redis的集群，可以使用红锁

6.红锁

  Redis Lock的简称Red Lock 

(1)原理

1.获取当前Unix时间，以ms为单位

2.轮流用相同的key和随机值在N个节点上请求锁，每个客户端在每个master上请求锁时，会有一个和总的锁释放时间小得多的超时时间，例如:如果锁的自动释放时间是10S，那么每个节点锁请求的超时时间可能是5-50ms，这可以防止一个客户端在某个宕机的master上阻塞过长时间，如果一个master节点不可用了，应该尽快尝试获取下一个master节点

3.客户端计算第二步获取锁花费时间，只有当客户端在大多数master节点上成功获取了锁（如果有5台redis，超过3台即可）,而且总消耗时间不超过锁的释放时间，这个锁就认为是获取成功了

4.如果锁获取成功了，那么现在自动释放时间是最初的锁释放时间减去之前获取没有获取成功的锁

5.如果锁获取失败了，不管是因为获取成功的锁不超过一半还是因为总消耗时间超过了锁释放时间，客户端都会到每个master节点上释放锁，即便是那些它认为没有获取成功的锁

（2）引入依赖

   <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.13.6</version>
        </dependency>

(3)配置redis

public class RedisConfig {
    @Autowired
    private  RedisSentinelProperties properties;
 
   @Bean(name="redissonRed1")
    public RedissonClient redissonRed1(){
       Config config = new Config();
       config.useSingleServer().setAddress("localhost:6379").setDatabase(0);
       return Redisson.create(config);
   }
    @Bean(name="redissonRed2")
    public RedissonClient redissonRed2(){
        Config config = new Config();
        config.useSingleServer().setAddress("localhost:6380").setDatabase(0);
        return Redisson.create(config);
    }
    @Bean(name="redissonRed3")
    public RedissonClient redissonRed3(){
        Config config = new Config();
        config.useSingleServer().setAddress("localhost:6381").setDatabase(0);
        return Redisson.create(config);
    }
    @Bean(name="redissonRed4")
    public RedissonClient redissonRed4(){
        Config config = new Config();
        config.useSingleServer().setAddress("localhost:6382").setDatabase(0);
        return Redisson.create(config);
    }
    @Bean(name="redissonRed5")
    public RedissonClient redissonRed5(){
        Config config = new Config();
        config.useSingleServer().setAddress("localhost:6383").setDatabase(0);
        return Redisson.create(config);
    }
 
 
 
}

配置了五台redis

(3)使用redlock

public class grabOrder {
 
   
    @Autowired
    @Qualifier("redissonRed1")
    private RedissonClient redissonRed1;
    @Autowired
    @Qualifier("redissonRed2")
    private RedissonClient redissonRed2;
    @Autowired
    @Qualifier("redissonRed3")
    private RedissonClient redissonRed3;
    @Autowired
    @Qualifier("redissonRed4")
    private RedissonClient redissonRed4;
    @Autowired
    @Qualifier("redissonRed5")
    private RedissonClient redissonRed5;
    
    
 
    public String fightOrder(int goodId,int userId){
 
        //生成key
        String  key = "goodsId_" + goodId+"";
 
        RLock rLock1 = redissonRed1.getLock(key);
        RLock rLock2 = redissonRed1.getLock(key);
        RLock rLock3 = redissonRed1.getLock(key);
        RLock rLock4 = redissonRed1.getLock(key);
        RLock rLock5 = redissonRed1.getLock(key);
 
        RedissonRedLock redLock = new RedissonRedLock(rLock1,rLock2,rLock3,rLock4,
                rLock5);
        
        
        try{
            //设置过期时间为30
            redLock.lock();
 
           boolean b = seckillOrderService.add(goodId,userId);
           if(b){
               System.out.println("用户:"+userId+"抢单成功！");
           }else {
               System.out.println("用户:"+userId+"抢单失败!");
           }
        }finally {
            redLock.unlock();
 
        }
        return null;
 
    }
 
}

7.redis分布式锁的问题

 当进程A从Redis中刚刚获取锁，此时系统进行完整垃圾回收（full fc）,进程全部停顿，STW，

这时，进程B可以从redis中获取锁，因为A的key已经过期了，这种情况可以使用ZooKeeper结合mysql乐观锁解决

四.Zookeper分布式锁

1.原理

(1)获取锁

在zookeeper中创建一个持久节点“good1”,当第一个客户获取锁时，在“good1”下面创建一个临时节点“good1 id-10001”,然后客户端1在“good1”下面查询所有的临时顺序节点并排序，判断自己创建的节点是不是序号最小的一个，如果是，则获取锁成功

 此时，如果有其他客户端2来获取锁，则在节点“good1”下面创建一个临时顺序节点good1 id-10002,客户端2再查找节点good1下面所有的临时节点并排序，判断自己的节点是不是最小的一个，如果不是，客户端2向比它靠前的节点id-10001注册Watcher,用于监听id-10001的状态，这表示客户端2抢锁失败，进入等待状态，依次类推客户端3,4

(2)释放锁

1.当任务完成时，客户端1会显示调用删除节点id-10001的命令

2.在执行任务，客户端崩溃，根据临时节点的特性，Zookeeper与客户端相关的节点会自己删除，避免了死锁

因为客户端2监听客户端1的状态，当id-10001被删除时，id-10002成为最小的节点，拿到了锁

2.Zookeepr结合mysql乐观锁

当进程1拿到锁时，将zookeeper节点的序号0保存在程序中，并且在Mysql中保存，当最终业务操作时，通过where条件判断mysql字段序号是否是0，如果是，则更新，不是就回滚

若进程1发送了STW，连接断开导致节点删除，此时进程2创建了节点，拿到了序号1，并且在程序中保存了1，然后更新了MYSQL中 的序号，如果进程1从STW恢复过来去执行业务的时候，发现MYSQL中的序号变成了1，则回滚

3.代码实现

（1）引入依赖

      <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.0</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.zookeeper</groupId>
                    <artifactId>zookeeper</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.14</version>
        </dependency>

（3）创建CuratorFrameWork

  
    public CuratorFramework curatorFramework(){
        
        ExponentialBackoffRetry retry = new ExponentialBackoffRetry(1000,3);
        //创建client
        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(
                "localhost:2181",retry
        );
        //添加watch监听器
        curatorFramework.getCuratorListenable().addListener(new
                MyCuratorListener());
        curatorFramework.start();
        return curatorFramework;
                
        
    }

(4)使用

 
    @Autowired
    private CuratorFramework client;
  
 
    public String fightOrder(int goodId,int userId){
 
        //生成key
       String lockPath = "/order"+goodId;
       
       InterProcessMutex lock = new InterProcessMutex(client,lockPath);
 
 
        try{
           if(lock.acquire(10, TimeUnit.HOURS)){
 
               boolean b = seckillOrderService.add(goodId,userId);
               if(b){
                   System.out.println("用户:"+userId+"抢单成功！");
               }else {
                   System.out.println("用户:"+userId+"抢单失败!");
               }
               
               
           }
 
          
        }finally {
           lock.release();
        }
        return null;
 
    }