实现Redis的分布式锁,除了自己基于redis client原生api来实现之外,还可以使用开源框架:Redission,Redisson是一个企业级的开源Redis Client,也提供了分布式锁的支持。
一、Redisson原理分析
(1)加锁机制
线程去获取锁,获取成功: 执行lua脚本,保存数据到redis数据库。
线程去获取锁,获取失败: 一直通过while循环尝试获取锁,获取成功后,执行lua脚本,保存数据到redis数据库。
如果该客户端面对的是一个redis cluster集群,他首先会根据hash节点选择一台机器,发送一段lua脚本到redis上.
lua脚本
Redisson源码中,执行redis命令的是lua脚本,其中主要用到如下几个概念。
-
- redis.call() 是执行redis命令.
- KEYS[1] 是指脚本中第1个参数
- ARGV[1] 是指脚本中第一个参数的值
- 返回值中nil与false同一个意思。
需要注意的是,在redis执行lua脚本时,相当于一个redis级别的锁,不能执行其他操作,类似于原子操作,也是redisson实现的一个关键点。
另外,如果lua脚本执行过程中出现了异常或者redis服务器直接宕掉了,执行redis的根据日志回复的命令,会将脚本中已经执行的命令在日志中删除。
(2)锁互斥机制
如果客户端2来尝试加锁,执行了同样的一段lua脚本,会咋样呢?很简单,第一个if判断会执行“exists myLock”,发现myLock这个锁key已经存在了。接着第二个if判断,判断一下,myLock锁key的hash数据结构中,是否包含客户端2的ID,但是明显不是的,因为那里包含的是客户端1的ID。
所以,客户端2会获取到pttl myLock返回的一个数字,这个数字代表了myLock这个锁key的**剩余生存时间。**比如还剩15000毫秒的生存时间。此时客户端2会进入一个while循环,不停的尝试加锁。
(3)watch dog自动延期机制
在一个分布式环境下,假如一个线程获得锁后,突然服务器宕机了,那么这个时候在一定时间后这个锁会自动释放,你也可以设置锁的有效时间(不设置默认30秒),这样的目的主要是防止死锁的发生。
只要客户端1一旦加锁成功,就会启动一个watch dog看门狗,他是一个后台线程,会每隔10秒检查一下,如果客户端1还持有锁key,那么就会不断的延长锁key的生存时间。
(4)可重入加锁机制
Redisson可以实现可重入加锁机制的原因,我觉得跟两点有关:
1、Redis存储锁的数据类型是 Hash类型
2、Hash数据类型的key值包含了当前线程信息。
下面是redis存储的数据
这里表面数据类型是Hash类型,Hash类型相当于我们java的 <key,<key1,value>>
类型,这里key是指 'redisson'
它的有效期还有9秒,我们再来看里们的key1值为078e44a3-5f95-4e24-b6aa-80684655a15a:45
它的组成是:
guid + 当前线程的ID。后面的value是就和可重入加锁有关。
(5)锁释放机制
如果执行lock.unlock(),就可以释放分布式锁,就是每次都对myLock数据结构中的那个加锁次数减1。如果发现加锁次数是0了,说明这个客户端已经不再持有锁了,此时就会用:“del myLock”命令,从redis里删除这个key。然后呢,另外的客户端2就可以尝试完成加锁了。这就是所谓的分布式锁的开源Redisson框架的实现机制。
一般我们在生产系统中,可以用Redisson框架提供的这个类库来基于redis进行分布式锁的加锁与释放锁。
Redis分布式锁的缺点
redis cluster,或者是redis master-slave架构的主从异步复制导致的redis分布式锁的最大缺陷:在redis master实例宕机的时候,可能导致多个客户端同时完成加锁。导致各种脏数据的产生。
二、基于redission的分布式锁实现
在分布式环境中,很多场景,如:秒杀、ID生成… 都需要分布式锁。分布式锁的实现,可以基于redis的setnx,zk的临时节点,redission框架
<dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> <version>3.2.3</version> </dependency>
工具类RedissionUtils
public class RedissionUtils { private static Logger logger = LoggerFactory.getLogger(RedissionUtils.class); private static RedissionUtils redisUtils; private RedissionUtils() { } /** * 提供单例模式 * * @return */ public static RedissionUtils getInstance() { if (redisUtils == null) synchronized (RedisUtils.class) { if (redisUtils == null) redisUtils = new RedissionUtils(); } return redisUtils; } /** * 使用config创建Redisson Redisson是用于连接Redis Server的基础类 * * @param config * @return */ public RedissonClient getRedisson(Config config) { RedissonClient redisson = Redisson.create(config); logger.info("成功连接Redis Server"); return redisson; } /** * 使用ip地址和端口创建Redisson * * @param ip * @param port * @return */ public RedissonClient getRedisson(String ip, String port) { Config config = new Config(); config.useSingleServer().setAddress(ip + ":" + port); RedissonClient redisson = Redisson.create(config); logger.info("成功连接Redis Server" + " " + "连接" + ip + ":" + port + "服务器"); return redisson; } /** * 关闭Redisson客户端连接 * * @param redisson */ public void closeRedisson(RedissonClient redisson) { redisson.shutdown(); logger.info("成功关闭Redis Client连接"); } /** * 获取字符串对象 * * @param redisson * @param objectName * @return */ public <T> RBucket<T> getRBucket(RedissonClient redisson, String objectName) { RBucket<T> bucket = redisson.getBucket(objectName); return bucket; } /** * 获取Map对象 * * @param redisson * @param objectName * @return */ public <K, V> RMap<K, V> getRMap(RedissonClient redisson, String objectName) { RMap<K, V> map = redisson.getMap(objectName); return map; } /** * 获取有序集合 * * @param redisson * @param objectName * @return */ public <V> RSortedSet<V> getRSortedSet(RedissonClient redisson, String objectName) { RSortedSet<V> sortedSet = redisson.getSortedSet(objectName); return sortedSet; } /** * 获取集合 * * @param redisson * @param objectName * @return */ public <V> RSet<V> getRSet(RedissonClient redisson, String objectName) { RSet<V> rSet = redisson.getSet(objectName); return rSet; } /** * 获取列表 * * @param redisson * @param objectName * @return */ public <V> RList<V> getRList(RedissonClient redisson, String objectName) { RList<V> rList = redisson.getList(objectName); return rList; } /** * 获取队列 * * @param redisson * @param objectName * @return */ public <V> RQueue<V> getRQueue(RedissonClient redisson, String objectName) { RQueue<V> rQueue = redisson.getQueue(objectName); return rQueue; } /** * 获取双端队列 * * @param redisson * @param objectName * @return */ public <V> RDeque<V> getRDeque(RedissonClient redisson, String objectName) { RDeque<V> rDeque = redisson.getDeque(objectName); return rDeque; } /** * 此方法不可用在Redisson 1.2 中 在1.2.2版本中 可用 * * @param redisson * @param objectName * @return */ /** * public <V> RBlockingQueue<V> getRBlockingQueue(RedissonClient * redisson,String objectName){ RBlockingQueue * rb=redisson.getBlockingQueue(objectName); return rb; } */ /** * 获取锁 * * @param redisson * @param objectName * @return */ public RLock getRLock(RedissonClient redisson, String objectName) { RLock rLock = redisson.getLock(objectName); return rLock; } /** * 获取原子数 * * @param redisson * @param objectName * @return */ public RAtomicLong getRAtomicLong(RedissonClient redisson, String objectName) { RAtomicLong rAtomicLong = redisson.getAtomicLong(objectName); return rAtomicLong; } /** * 获取记数锁 * * @param redisson * @param objectName * @return */ public RCountDownLatch getRCountDownLatch(RedissonClient redisson, String objectName) { RCountDownLatch rCountDownLatch = redisson .getCountDownLatch(objectName); return rCountDownLatch; } /** * 获取消息的Topic * * @param redisson * @param objectName * @return */ public <M> RTopic<M> getRTopic(RedissonClient redisson, String objectName) { RTopic<M> rTopic = redisson.getTopic(objectName); return rTopic; } }
redission连接模式
//单机 RedissonClient redisson = Redisson.create(); Config config = new Config(); config.useSingleServer().setAddress("myredisserver:6379"); RedissonClient redisson = Redisson.create(config); //主从 Config config = new Config(); config.useMasterSlaveServers() .setMasterAddress("127.0.0.1:6379") .addSlaveAddress("127.0.0.1:6389", "127.0.0.1:6332", "127.0.0.1:6419") .addSlaveAddress("127.0.0.1:6399"); RedissonClient redisson = Redisson.create(config); //哨兵 Config config = new Config(); config.useSentinelServers() .setMasterName("mymaster") .addSentinelAddress("127.0.0.1:26389", "127.0.0.1:26379") .addSentinelAddress("127.0.0.1:26319"); RedissonClient redisson = Redisson.create(config); //集群 Config config = new Config(); config.useClusterServers() .setScanInterval(2000) // cluster state scan interval in milliseconds .addNodeAddress("127.0.0.1:7000", "127.0.0.1:7001") .addNodeAddress("127.0.0.1:7002"); RedissonClient redisson = Redisson.create(config);
三、redission的各种分布式锁
1)可重入锁:
Redisson的分布式可重入锁RLock,实现了java.util.concurrent.locks.Lock接口,以及支持自动过期解锁。同时还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。
// 最常见的使用方法 RLock lock = redisson.getLock("anyLock"); lock.lock(); //... lock.unlock(); //另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。 // 加锁以后10秒钟自动解锁 // 无需调用unlock方法手动解锁 lock.lock(10, TimeUnit.SECONDS); // 尝试加锁,最多等待100秒,上锁以后10秒自动解锁 boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS); if (res) { try { ... } finally { lock.unlock(); } }
大家都知道,如果负责储存这个分布式锁的Redisson节点宕机以后,而且这个锁正好处于锁住的状态时,这个锁会出现锁死的状态。为了避免这种情况的发生,Redisson内部提供了一个监控锁的看门狗,它的作用是在Redisson实例被关闭前,不断的延长锁的有效期。默认情况下,看门狗的检查锁的超时时间是30秒钟,也可以通过修改Config.lockWatchdogTimeout来另行指定。
Redisson同时还为分布式锁提供了异步执行的相关方法:
RLock lock = redisson.getLock("anyLock"); lock.lockAsync(); lock.lockAsync(10, TimeUnit.SECONDS); Future<Boolean> res = lock.tryLockAsync(100, 10, TimeUnit.SECONDS);
RLock对象完全符合Java的Lock规范。也就是说只有拥有锁的进程才能解锁,其他进程解锁则会抛出IllegalMonitorStateException错误。
2)公平锁(Fair Lock):
它保证了当多个Redisson客户端线程同时请求加锁时,优先分配给先发出请求的线程。所有请求线程会在一个队列中排队,当某个线程出现宕机时,Redisson会等待5秒后继续下一个线程,也就是说如果前面有5个线程都处于等待状态,那么后面的线程会等待至少25秒。使用方式同上,获取的时候使用如下方法:
RLock fairLock = redisson.getFairLock("anyLock");
3)联锁(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); // 给lock1,lock2,lock3加锁,如果没有手动解开的话,10秒钟后将会自动解开 lock.lock(10, TimeUnit.SECONDS); // 为加锁等待100秒时间,并在加锁成功10秒钟后自动解开 boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS); ... lock.unlock();
4)红锁(RedLock):
基于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); // 给lock1,lock2,lock3加锁,如果没有手动解开的话,10秒钟后将会自动解开 lock.lock(10, TimeUnit.SECONDS); // 为加锁等待100秒时间,并在加锁成功10秒钟后自动解开 boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS); ... lock.unlock();
5)读写锁(ReadWriteLock):
基于Redis的Redisson分布式可重入读写锁RReadWriteLock Java对象实现了java.util.concurrent.locks.ReadWriteLock接口。其中读锁和写锁都继承了RLock接口。分布式可重入读写锁允许同时有多个读锁和一个写锁处于加锁状态。
RReadWriteLock rwlock = redisson.getReadWriteLock("anyRWLock"); // 最常见的使用方法 rwlock.readLock().lock(); // 或 rwlock.writeLock().lock(); //另外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();
6)信号量(Semaphore):
基于Redis的Redisson的分布式信号量(Semaphore)Java对象RSemaphore采用了与java.util.concurrent.Semaphore相似的接口和用法。同时还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。
RSemaphore semaphore = redisson.getSemaphore("semaphore"); semaphore.acquire(); //或 semaphore.acquireAsync(); semaphore.acquire(23); semaphore.tryAcquire(); //或 semaphore.tryAcquireAsync(); semaphore.tryAcquire(23, TimeUnit.SECONDS); //或 semaphore.tryAcquireAsync(23, TimeUnit.SECONDS); semaphore.release(10); semaphore.release(); //或 semaphore.releaseAsync();
7)可过期性信号量(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);
8)门闩:
基于Redisson的Redisson分布式闭锁(CountDownLatch)Java对象RCountDownLatch采用了与java.util.concurrent.CountDownLatch相似的接口和用法。
RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch"); latch.trySetCount(1); latch.await(); // 在其他线程或其他JVM里 RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch"); latch.countDown();
9)分布式AtomicLong:
RAtomicLong atomicLong = redisson.getAtomicLong("myAtomicLong"); atomicLong.set(3); atomicLong.incrementAndGet(); atomicLong.get()
10)分布式BitSet:
RBitSet set = redisson.getBitSet("simpleBitset"); set.set(0, true); set.set(1812, false); set.clear(0); set.addAsync("e"); set.xor("anotherBitset");
11)分布式Object:
RBucket<AnyObject> bucket = redisson.getBucket("anyObject"); bucket.set(new AnyObject(1)); AnyObject obj = bucket.get(); bucket.trySet(new AnyObject(3)); bucket.compareAndSet(new AnyObject(4), new AnyObject(5)); bucket.getAndSet(new AnyObject(6));
12)分布式Set:
RSet<SomeObject> set = redisson.getSet("anySet"); set.add(new SomeObject()); set.remove(new SomeObject());
13)分布式List:
RList<SomeObject> list = redisson.getList("anyList"); list.add(new SomeObject()); list.get(0); list.remove(new SomeObject());
14)分布式Blocking Queue:
RBlockingQueue<SomeObject> queue = redisson.getBlockingQueue("anyQueue"); queue.offer(new SomeObject()); SomeObject obj = queue.peek(); SomeObject someObj = queue.poll(); SomeObject ob = queue.poll(10, TimeUnit.MINUTES);
15)分布式Map:
RMap<String, SomeObject> map = redisson.getMap("anyMap"); SomeObject prevObject = map.put("123", new SomeObject()); SomeObject currentObject = map.putIfAbsent("323", new SomeObject()); SomeObject obj = map.remove("123"); map.fastPut("321", new SomeObject()); map.fastRemove("321"); Future<SomeObject> putAsyncFuture = map.putAsync("321"); Future<Void> fastPutAsyncFuture = map.fastPutAsync("321"); map.fastPutAsync("321", new SomeObject()); map.fastRemoveAsync("321");
除此之外,还支持Multimap。
16)Map eviction:
现在Redis没有过期清空Map中的某个entry的功能,只能是清空Map所有的entry。Redission提供了这种功能。
RMapCache<String, SomeObject> map = redisson.getMapCache("anyMap"); // ttl = 10 minutes, map.put("key1", new SomeObject(), 10, TimeUnit.MINUTES); // ttl = 10 minutes, maxIdleTime = 10 seconds map.put("key1", new SomeObject(), 10, TimeUnit.MINUTES, 10, TimeUnit.SECONDS); // ttl = 3 seconds map.putIfAbsent("key2", new SomeObject(), 3, TimeUnit.SECONDS); // ttl = 40 seconds, maxIdleTime = 10 seconds map.putIfAbsent("key2", new SomeObject(), 40, TimeUnit.SECONDS, 10, TimeUnit.SECONDS);