基于redission的分布式锁

在分布式环境中，很多场景，如：秒杀、ID生成... 都需要分布式锁。分布式锁的实现，可以基于redis的setnx，zk的临时节点。。。今天我们介绍一种redis官方推荐的方法——redission。

1、pom.xml


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

2、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; 
    } 
    
 
    /** 
     
     * 
     * @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" + "\t" + "连接" + 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);

3、基于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);