济宁网站建设 果壳科技,wordpress制作数据排行榜,成都市区必去的景点,网站开发 组织架构引言 什么是分布式锁#xff1f; 分布式锁是分布式系统中用于控制多个进程或线程对共享资源的访问的一种机制。在分布式系统中#xff0c;由于存在多个服务实例或节点#xff0c;它们可能会同时尝试访问或修改同一份数据或资源。如果没有适当的同步机制#xff0c;就可能导…引言 什么是分布式锁 分布式锁是分布式系统中用于控制多个进程或线程对共享资源的访问的一种机制。在分布式系统中由于存在多个服务实例或节点它们可能会同时尝试访问或修改同一份数据或资源。如果没有适当的同步机制就可能导致数据不一致、重复处理或丢失更新等问题。分布式锁就是为了解决这些问题而设计的。 为什么要分布式锁  在单进程启动一个jvm的系统中当存在多个线程可以同时改变某个变量可变共享变量时就需要对变量或代码块做同步使其在修改这种变量时能够线性执行消除并发修改变量。而同步的本质是通过锁来实现的。为了实现多个线程在一个时刻同一个代码块只能有一个线程可执行那么需要在某个地方做个标记这个标记必须每个线程都能看到当标记不存在时可以设置该标记其余后续线程发现已经有标记了则等待拥有标记的线程结束同步代码块取消标记后再去尝试设置标记。这个标记可以理解为锁。 并发问题 解决方案 在java中可以通过synchronized和lock等手段来实现。  分布式锁 很多时候我们需要保证一个方法在同一时间内只能被同一个线程执行。在单机环境中通过 Java 提供的并发 API 我们可以解决但是在分布式环境下就没有那么简单啦。 分布式与单机情况下最大的不同在于其不是多线程而是多进程。多线程由于可以共享堆内存因此可以简单的采取内存作为标记存储位置。而进程之间甚至可能都不在同一台物理机上因此需要将标记存储在一个所有进程都能看到的地方。 那么原来的方案就不行了 如果是在集群或分布式环境中要保证多进程中的多线程的线程安全就要使用分布式锁分布式锁的目的就是在分布式/集群环境中使用加锁手段保证多个服务节点对同一个数据进行顺序操作保证数据的安全性如上图多个服务都在同时扣减库存我们需要对减库存进行顺序操作如  其实实现分布式锁的原理也很简单就是需要得有一把唯一且共享的锁多个服务同时去获取锁但是只有一个服务才能获取到锁其他没有获取到锁的服务需要等待或者自旋等获取到锁的服务业务执行完成释放锁其他的服务就可以再次尝试获取锁。  Redis分布式锁的原理 实现分布式锁的方案有很多比如基于数据库实现分布式锁使用ZooKeeper实现分布式锁本文采用的是使用Redis实现分布式锁方案。 加锁和释放锁Redis提供了一个命令setnx 可以来实现分布式锁该命令只在键 key 不存在的情况下 将键 key 的值设置为 value 若键 key 已经存在 则 SETNX 命令不做任何动作。根据这一特性我们就可以制定Redis实现分布式锁的方案了。 简单理解就是 如果三个服务同时抢锁服务A抢先一步执行setnx(lock_stock,1)加上锁那么当服务B在执行setnx(lock_stock,1)加锁的时候就会失败服务C也一样服务A抢到锁执行完业务逻辑后就会释放锁可以使用del(lock_stock)删除锁其他服务就可以执行setnx(lock_stock,1)加锁了如图 锁超时问题  这里有一个问题如果获取到锁的服务在释放锁的时候宕机了那么Redis中lock-stock不就永远存在那锁不就释放不了么别的服务也就没办法获取到锁就造成了死锁为了解决这个问题我们需要设置锁的自动超时也就是Key的超时自动删除即使服务宕机没有调用del释放锁那么锁本身也有超时时间可以自动删除锁别的服务就可以获取锁了Redis中Key的过期时间可以使用Redis的 expirelock_stock30命令实现这里给出伪代码如下 ifjedis.setnxlock_stock1 1{ //获取锁expirelock_stock5 //设置锁超时try {业务代码} finally {jedis.dellock_stock //释放锁} }原子性问题 上面的代码依然有问题就是setnx获取锁和expire不是原子性操作假设有一极端情况当线程通过setnxlock_stock1获取到锁还没来得及执行expirelock_stock30设置锁的过期时间服务就宕机了那是不是锁也永远得不到释放呢又变成了死锁这个问题可以使用set命令解决我们先来看一下这个命令的语法 SET key value [EX seconds] [PX milliseconds] [NX|XX]从 Redis 2.6.12 版本开始 SET 命令的行为可以通过一系列参数来修改 EX seconds 将键的过期时间设置为 seconds 秒。 执行 SET key value EX seconds 的效果等同于执行 SETEX key seconds value 。PX milliseconds 将键的过期时间设置为 milliseconds 毫秒。 执行 SET key value PX milliseconds 的效果等同于执行 PSETEX key milliseconds value 。NX 只在键不存在时 才对键进行设置操作。 执行 SET key value NX 的效果等同于执行 SETNX key value 。XX 只在键已经存在时 才对键进行设置操作。 也就是说该命令可以当做setnx和expire的组合命令来使用而且是原子性的改造代码如下 ifset(lock_stock,1,NX,EX,5) 1{ //获取锁并设置超时try {业务代码} finally {dellock_stock //释放锁} }锁的误删除问题 上面的方案依然有问题就是在del释放锁的时候可能会误删除别人加的锁例如服务A获取到锁lock_stock过期时间为 5s如果在服务A执行业务逻辑的这一段时间内锁到期自动删除且别的服务获取到了锁lock_stock那么服务A业务执行完成执行del(lock_stock)是不是会把别人的锁给删除掉呢如图 那么这个问题怎么解决呢我们可以在删除锁的时候先判断一下要删除的锁是不是自己上的锁比如可以把锁的值使用一个UUID在释放锁的时候先获取一下锁的值和当前业务中创建的UUID是不是同一个如果是才执行·del删除锁当然也可以使用线程的ID替代UUID代码如下 String uuid UUID.randomUUID().toString(); ifjedis.set(lock_stock,uuid,NX,EX,5) 1{ //获取锁并设置超时try {业务代码} finally {String lockValue jedis.get(lock_stock); //获取锁的值if(lockValue.equals(uuid)){ //判断是不是自己的锁jedis.dellock_stock //释放锁}} }Lua脚本保证原子性 但是上面的代码依然有问题就是判断锁的代码和删除锁的代码也不是原子性的依然可能会导致锁的误删除问题比如服务A在判断锁成功准备删除锁时锁自动过期别的服务B获取到了锁然后服务A执行DEL就可能会把服务B的锁给删除掉所以我们必须保证 获取锁 - 判断锁 - 删除锁 的操作是原子性的才可以解决方案可以使用RedisLua脚本来解决一致性问题 String script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end;这是一段Lua脚本可以保证多个命令的原子性 redis.call(‘get’, KEYS[1]) 是调用redis的get命令key可以通过参数传入 ARGV[1] 意思是是否和 某个值相等这里的值也可以参数传入then return redis.call(‘del’, KEYS[1]) 如果相等就执行 redis.call(del, KEYS[1]) 删除操作else return 0 end 否则就返回 0 如果我们把数据带入KEYS[1]的值为“lock_stock”,ARGV[1]的值为UUID如“xoxoxo”所以大概的含义是如果调用get(“lock_stock”)获取到的值 等于 “xoxoxo” ,那就调用 del(“lock_stock”)否则就返回 0 。 说白了就是把我们上面的判断锁和删除锁的动作使用Lua脚本去执行而已现在代码可以这样写了 String uuid UUID.randomUUID().toString(); ifjedis.set(lock_stock,uuid,NX,EX,5) 1{ //获取锁并设置超时try {业务代码} finally {//lua脚本String script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end;//执行脚本jedis.eval(script, Collections.singletonList(lock_stock),Collections.singletonList(uuid));} }Arrays.asList(“lock_stock”) 转给 KEYS[1]Arrays.asList(uuid)转给 ARGV[1] 可重入锁 上面的代码是不完整的如果某个线程没有获取到锁是不是就不会进入 IF 呢如果是这样的话未获取到锁的线程就执行失败了啥也没做这是不可行的我们是不是需要让未获取到锁的线程等待片刻之后再次尝试获取锁呢如下 public void method(){String uuid UUID.randomUUID().toString();ifjedis.set(lock_stock,uuid,NX,EX,5) 1{ //获取锁并设置超时try {业务代码} finally {//lua脚本String script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end;//执行脚本jedis.eval(script, Collections.singletonList(lock_stock),Collections.singletonList(uuid));}}else{//休眠一会儿重入方法尝试获取锁Thread.sleep(100);method(); //自旋重新进入方法} } 上面的代码增加了else获取锁失败的逻辑休眠一会儿后重入方法尝试重新获取锁休眠时间结合业务逻辑的执行时间设定 分布式锁的特点 当然要实现一个分布式锁还需要考虑一些东西比如Redis的健壮性它不能随便挂掉这里总结一下分布式锁的一些要素 首先为了确保分布式锁可用我们至少要确保锁的实现同时满足以下四个条件 互斥性同一时间只能一个节点获取到锁其他节点需要等待获取到锁的节点释放了锁才可以获取到锁而这里的等待一般是通过阻塞和自旋两种方式安全性解铃还须系铃人只能释放自己的锁不能误删别人的锁死锁比如在节点宕机时最容易出现锁没被释放的问题然后出现死锁所以做锁的过期容错当Redis宕机客户端仍然可以释放锁可重入获取锁失败可以重新尝试获取锁 要实现一个分布式锁是不是要考虑很多细节呢其实不用做什么麻烦我们有更专业的工具已经帮我们封装好上面的所有细节 Redisson的实现分布式锁 Redisson是什么 我们操作Redis的手段有很多在Java中可以使用Jedis或者Redisson本文章在于讨论Redisson是如何操作Java的下面是对Redisson的概述官方文档 Redisson是一个实现的Java操作Redis的工具包它不仅提供了一系列常用的操作Redis的API还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最简单和最便捷的方法Redisson的宗旨是促进使用者对Redis的关注分离从而让使用者能够将精力更集中地放在处理业务逻辑上。 Redisson的集成 导入依赖 !-- https://mvnrepository.com/artifact/org.redisson/redisson -- dependencygroupIdorg.redisson/groupIdartifactIdredisson/artifactIdversion3.13.6/version /dependency 配置一个单机Redis Configuration public class RedissonConfig {//创建客户端Beanpublic RedissonClient redissonClient(){Config config new Config();config.useSingleServer().setAddress(redis://127.0.0.1:6379);//.setPassword(123456);return Redisson.create(config);} }Redisson实现分布式锁 官方对分布式锁的定义 大家都知道如果负责储存这个分布式锁的Redisson节点宕机以后而且这个锁正好处于锁住的状态时这个锁会出现锁死的状态。为了避免这种情况的发生Redisson内部提供了一个监控锁的看门狗它的作用是在Redisson实例被关闭前不断的延长锁的有效期。默认情况下看门狗的检查锁的超时时间是30秒钟也可以通过修改Config.lockWatchdogTimeout来另行指定。另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。 以上是Redisson官方文档对分布式锁的解释总结下来有两点 Redisson加锁自动有过期时间30s监控锁的看门狗发现业务没执行完会自动进行锁的续期(重回30s)这样做的好处是防止在程序执行期间锁自动过期被删除问题当业务执行完成不再给锁续期即使没有手动释放锁锁的过期时间到了也会自动释放锁 可重入锁Reentrant Lock  基于Redis的Redisson分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口。同时还提供了异步Async、反射式Reactive和RxJava2标准的接口 一个简单的锁分布式锁案例如下 Autowiredprivate RedissonClient redissonClient;Testpublic void testLock1(){RLock rLock redissonClient.getLock(lock_stock);rLock.lock(); //阻塞式等待过期时间30stry{System.out.println(加锁成功....);System.out.println(执行业务....);}finally {rLock.unlock();System.out.println(释放锁....);}}另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了如下 Testpublic void testLock2(){RLock rLock redissonClient.getLock(lock_stock);// 加锁以后10秒钟自动解锁// 无需调用unlock方法手动解锁rLock.lock(10, TimeUnit.SECONDS);try{System.out.println(加锁成功....);System.out.println(执行业务....);}finally {rLock.unlock();System.out.println(释放锁....);}}Redisson对分布式锁实现细节进行了封装帮我们处理了分布式锁面临的一些列问题那么Redisson是如何工作的呢 如果没有设置过期时间Redisson以 30s 作为锁的默认过期时间获取锁成功后(底层也用到了Lua脚本保证原子性)会开启一个定时任务定时进行锁过期时间续约即每次都把过期时间设置成 30s定时任务 10s执行一次(看门狗)如果设置了过期时间直接把设定的过期时间作为锁的过期时间然后使用Lua脚本获取锁没获取到锁的线程会while自旋重入不停地尝试获取锁 这里需要注意rLock.lock(10, TimeUnit.SECONDS)指定了解锁时间Redisson就不会再自动续期那么如果在线程A业务还没执行完就自动解锁了这时候线程B获取到锁继续执行业务那么等线程A业务执行完释放锁就可能会把线程B的锁删除当然这种情况Redisson会报异常但是这种情况是没有把所有线程都锁住的所以如果要手动设定过期时间需要让过期时间比业务逻辑执行的时间长才对。 Redisson同时还为分布式锁提供了异步执行的相关方法 Test public void testLock3() {RLock rLock redissonClient.getLock(lock_stock);try{//rLock.lockAsync();//10秒自动释放锁//rLock.lockAsync(10, TimeUnit.SECONDS);//尝试加锁等待2秒上锁以后10秒自动释放锁FutureBoolean res rLock.tryLockAsync(2, 10, TimeUnit.SECONDS);if(res.get()){System.out.println(加锁成功....);System.out.println(执行业务....);}} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}finally {rLock.unlock();System.out.println(释放锁....);} }RLock对象完全符合Java的Lock规范。也就是说只有拥有锁的进程才能解锁其他进程解锁则会抛出IllegalMonitorStateException错误。但是如果遇到需要其他进程也能解锁的情况请使用分布式信号量Semaphore 对象。 公平锁Fair Lock 基于Redis的Redisson分布式可重入公平锁也是实现了java.util.concurrent.locks.Lock接口的一种RLock对象。同时还提供了异步Async、反射式Reactive和RxJava2标准的接口。它保证了当多个Redisson客户端线程同时请求加锁时优先分配给先发出请求的线程。所有请求线程会在一个队列中排队当某个线程出现宕机时Redisson会等待5秒后继续下一个线程也就是说如果前面有5个线程都处于等待状态那么后面的线程会等待至少25秒 Testpublic void testLock5() {RLock fairLock redissonClient.getFairLock(anyLock);try{// 最常见的使用方法fairLock.lock();}finally {fairLock.unlock();System.out.println(释放锁....);}}另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了 // 10秒钟以后自动解锁 // 无需调用unlock方法手动解锁 fairLock.lock(10, TimeUnit.SECONDS);// 尝试加锁最多等待100秒上锁以后10秒自动解锁 boolean res fairLock.tryLock(100, 10, TimeUnit.SECONDS); ... fairLock.unlock();Redisson同时还为分布式可重入公平锁提供了异步执行的相关方法 RLock fairLock redisson.getFairLock(anyLock); fairLock.lockAsync(); fairLock.lockAsync(10, TimeUnit.SECONDS); FutureBoolean res fairLock.tryLockAsync(100, 10, TimeUnit.SECONDS);联锁MultiLock 基于Redis的Redisson分布式联锁RedissonMultiLock对象可以将多个RLock对象关联为一个联锁每个RLock对象实例可以来自于不同的Redisson实例。 RLock lock1 redissonInstance1.getLock(lock1); RLock lock2 redissonInstance2.getLock(lock2); RLock lock3 redissonInstance3.getLock(lock3);RedissonMultiLock lock new RedissonMultiLock(lock1, lock2, lock3); // 同时加锁lock1 lock2 lock3 // 所有的锁都上锁成功才算成功。 lock.lock(); ... lock.unlock();另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。 RedissonMultiLock lock new RedissonMultiLock(lock1, lock2, lock3); // 给lock1lock2lock3加锁如果没有手动解开的话10秒钟后将会自动解开 lock.lock(10, TimeUnit.SECONDS);// 为加锁等待100秒时间并在加锁成功10秒钟后自动解开 boolean res lock.tryLock(100, 10, TimeUnit.SECONDS); ... lock.unlock();红锁RedLock Redis常用的方式有单节点、主从模式、哨兵模式、集群模式在后三种模式中可能会出现 异步数据丢失脑裂问题Redis官方提供了解决方案RedLock,RedLock是基于redis实现的分布式 锁它能够保证以下特性 容错性只要多数节点的redis实例正常运行就能够对外提供服务,加锁释放锁互斥性只能有一个客户端能获取锁即使发生了网络分区或者客户端宕机也不会发生死锁 基于Redis的Redisson红锁RedissonRedLock对象实现了Redlock介绍的加锁算法。该对象也可以用来将多个RLock对象关联为一个红锁每个RLock对象实例可以来自于不同的Redisson实例。 RLock lock1 redissonInstance1.getLock(lock1); RLock lock2 redissonInstance2.getLock(lock2); RLock lock3 redissonInstance3.getLock(lock3);RedissonRedLock lock new RedissonRedLock(lock1, lock2, lock3); // 同时加锁lock1 lock2 lock3 // 红锁在大部分节点上加锁成功就算成功。 lock.lock(); ... lock.unlock();另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。 RedissonRedLock lock new RedissonRedLock(lock1, lock2, lock3); // 给lock1lock2lock3加锁如果没有手动解开的话10秒钟后将会自动解开 lock.lock(10, TimeUnit.SECONDS);// 为加锁等待100秒时间并在加锁成功10秒钟后自动解开 boolean res lock.tryLock(100, 10, TimeUnit.SECONDS); ... lock.unlock();读写锁ReadWriteLock 基于Redis的Redisson分布式可重入读写锁RReadWriteLock Java对象实现了java.util.concurrent.locks.ReadWriteLock接口。其中读锁和写锁都继承了RLock接口。 分布式可重入读写锁允许同时有多个读锁和一个写锁处于加锁状态,即使用同一个RReadWriteLock加写锁和读锁多个读锁是需要等待写释放锁才能加锁成功如下 Testpublic void testWriteLock() {//获取读写锁RReadWriteLock readWriteLock redissonClient.getReadWriteLock(ReadWriteLock);//获取写锁RLock rLock readWriteLock.writeLock();try{//加上写锁读会等待rLock.lock();System.out.println(写锁加锁成功);Thread.sleep(200000);System.out.println(处理写业务...);} catch (InterruptedException e) {e.printStackTrace();}finally {rLock.unlock();System.out.println(释放写锁....);}}Testpublic void testReadLock() {//获取读写锁RReadWriteLock readWriteLock redissonClient.getReadWriteLock(ReadWriteLock);//获取读锁RLock rLock readWriteLock.readLock();try{//加上读锁如果写锁没释放会等待rLock.lock();System.out.println(读锁加锁成功);System.out.println(处理读业务...);}finally {rLock.unlock();System.out.println(释放读锁....);}}如果 testWriteLock 写方法先自行先加上写锁 那么 testReadLock读方法中的加锁代码会等待直到写锁释放。 当然如果多个线程全是读锁没有写锁那相当于是没有加锁不会等待其他情况只要有写锁参与后执行加锁的线程都要等先执行加锁的线程释放锁不管是先读还是先写又或者是写和写。这种锁能够保证读锁能读到的数据始终是写完之后的数据。 另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。 // 10秒钟以后自动解锁 // 无需调用unlock方法手动解锁 rwlock.readLock().lock(10, TimeUnit.SECONDS); // 或 rwlock.writeLock().lock(10, TimeUnit.SECONDS);// 尝试加锁最多等待100秒上锁以后10秒自动解锁 boolean res rwlock.readLock().tryLock(100, 10, TimeUnit.SECONDS); // 或 boolean res rwlock.writeLock().tryLock(100, 10, TimeUnit.SECONDS); ... lock.unlock();信号量Semaphore 基于Redis的Redisson的分布式信号量SemaphoreJava对象RSemaphore采用了与java.util.concurrent.Semaphore相似的接口和用法。同时还提供了异步Async、反射式Reactive和RxJava2标准的接口。 信号量可以看做是在Redis中保存了一个数字然后可以实现原子性的加或者减比如说有一商品需要拿100个做秒杀我们就可以把这个库存数量做成信号量然后实现原子性加减操作 Test public void testReadLock5() throws InterruptedException {//获得到一个信号量RSemaphore semaphore redissonClient.getSemaphore(semaphore);//设置信号量的值boolean setPermits semaphore.trySetPermits(1000);System.out.println(setPermits);System.out.println(可用数量semaphore.availablePermits());} Test public void testReadLock6() throws InterruptedException {//获得到一个信号量RSemaphore semaphore redissonClient.getSemaphore(semaphore);//获取 2 个信号量 值会减去 2 如果获取不到方法会阻塞semaphore.acquire(2);System.out.println(可用数量semaphore.availablePermits());//尝试获取 2 个信号量 值会减去 2 如果获取不到方法不会boolean tryAccquireSuccess semaphore.tryAcquire(2);System.out.println(tryAccquireSuccess);System.out.println(可用数量semaphore.availablePermits());}Test public void testReadLock7() throws InterruptedException {//获得到一个信号量RSemaphore semaphore redissonClient.getSemaphore(semaphore);//释放2个值数量会加回去semaphore.release(2);System.out.println(可用数量semaphore.availablePermits()); }可过期性信号量PermitExpirableSemaphore 基于Redis的Redisson可过期性信号量PermitExpirableSemaphore是在RSemaphore对象的基础上为每个信号增加了一个过期时间。每个信号可以通过独立的ID来辨识释放时只能通过提交这个ID才能释放。它提供了异步Async、反射式Reactive和RxJava2标准的接口。 RPermitExpirableSemaphore semaphore redisson.getPermitExpirableSemaphore(mySemaphore); String permitId semaphore.acquire(); // 获取一个信号有效期只有2秒钟。 String permitId semaphore.acquire(2, TimeUnit.SECONDS); // ... semaphore.release(permitId);闭锁CountDownLatch 基于Redisson的Redisson分布式闭锁CountDownLatchJava对象RCountDownLatch采用了与java.util.concurrent.CountDownLatch相似的接口和用法。 闭锁可以实现多个线程都执行完才是完成的效果否则闭锁会等待。 RCountDownLatch latch redisson.getCountDownLatch(anyCountDownLatch); //设置2个数量 latch.trySetCount(2);//await方法会等待等待其他线程 countDown 完成所有的trySetCount(2)次就结束闭锁 latch.await();// 在其他线程或其他JVM里 RCountDownLatch latch redisson.getCountDownLatch(anyCountDownLatch); //完成第1个 latch.countDown();// 在其他线程或其他JVM里 RCountDownLatch latch redisson.getCountDownLatch(anyCountDownLatch); //完成第2个 闭锁完成 latch.countDown(); 文章结束啦如果对你有帮助请一定给个好评哦~~