一、redisson介绍
redisson实现了分布式和可扩展的java数据结构，支持的数据结构有：List, Set, Map, Queue, SortedSet, ConcureentMap, Lock, AtomicLong, CountDownLatch。并且是线程安全的，底层使用Netty 4实现网络通信。和jedis相比，功能比较简单，不支持排序，事务，管道，分区等redis特性，可以认为是jedis的补充，不能替换jedis。
二、redisson几种锁介绍
1、可重入锁（Reentrant Lock）
基于Redis的Redisson分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口
RLock lock = redisson.getLock("anyLock");
// 最常见的使用方法
lock.lock();
另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。
// 加锁以后10秒钟自动解锁
// 无需调用unlock方法手动解锁
lock.lock(10, TimeUnit.SECONDS); // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
...
lock.unlock();
Redisson同时还为分布式锁提供了异步执行的相关方法：
RLock lock = redisson.getLock("anyLock");
lock.lockAsync();
lock.lockAsync(10, TimeUnit.SECONDS);
Future<Boolean> res = lock.tryLockAsync(100, 10, TimeUnit.SECONDS);
2.公平锁（Fair Lock）
基于Redis的Redisson分布式可重入公平锁也是实现了java.util.concurrent.locks.Lock接口的一种RLock对象。它保证了当多个Redisson客户端线程同时请求加锁时，优先分配给先发出请求的线程。
RLock fairLock = redisson.getFairLock("anyLock");
// 最常见的使用方法
fairLock.lock();
同样的，Fair lock也提供加锁时间
// 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);
Future<Boolean> res = fairLock.tryLockAsync(100, 10, TimeUnit.SECONDS);
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.红锁（Red Lock）
基于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接口。同时还支持自动过期解锁。该对象允许同时有多个读取锁，但是最多只能有一个写入锁。
RReadWriteLock rwlock = redisson.getLock("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. 闭锁（CountDownLatch）
基于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();
三、redisson分布式锁在业务中的应用
1、引入相关pom
<!-- redisson -->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.5.0</version>
</dependency>
2、将redisson交由spring管理，本文采用redisson的集群模式装配，另外可配置哨兵模式，单点等
/**
* redisson客户端参数配置类
*/
@Configuration
@Data
public class RedissonProperties { private int idleConnectionTimeout = 10000; private int pingTimeout = 1000; private int connectTimeout = 10000; private int timeout = 3000; private int retryAttempts = 3; private int retryInterval = 1500; private int reconnectionTimeout = 3000; private int failedAttempts = 3; private int subscriptionsPerConnection = 5; private String clientName = "none"; private int subscriptionConnectionMinimumIdleSize = 64; private int subscriptionConnectionPoolSize = 256; private int slaveConnectionMinimumIdleSize = 64; private int slaveConnectionPoolSize = 256; private int masterConnectionMinimumIdleSize = 64; private int masterConnectionPoolSize = 256; private ReadMode readMode = ReadMode.MASTER; private SubscriptionMode subscriptionMode = SubscriptionMode.MASTER; private int scanInterval = 1000; @Value("${rediscluster.pwd}")
private String password; @Value("${redis.cluster}")
private String nodeAddress; @Value("${redis.cluster1}")
private String nodeAddress1; @Value("${redis.cluster2}")
private String nodeAddress2; }
/**
* 初始化redisson Bean
*
* @author LiJunJun
* @date 2018/10/19
*/
@Configuration
public class RedissonAutoConfiguration { @Autowired
private RedissonProperties redssionProperties; /**
* 集群模式自动装配
*
* @return
*/
@Bean
public RedissonClient redissonClient() { Config config = new Config();
String passWord = redssionProperties.getPassword();
ClusterServersConfig serverConfig = config.useClusterServers();
serverConfig.addNodeAddress(redssionProperties.getNodeAddress(), redssionProperties.getNodeAddress1(), redssionProperties.getNodeAddress2());
serverConfig.setPingTimeout(redssionProperties.getPingTimeout());
serverConfig.setConnectTimeout(redssionProperties.getConnectTimeout());
serverConfig.setTimeout(redssionProperties.getTimeout());
serverConfig.setRetryAttempts(redssionProperties.getRetryAttempts());
serverConfig.setRetryInterval(redssionProperties.getRetryInterval());
serverConfig.setReconnectionTimeout(redssionProperties.getReconnectionTimeout());
serverConfig.setFailedAttempts(redssionProperties.getFailedAttempts());
serverConfig.setSubscriptionsPerConnection(redssionProperties.getSubscriptionsPerConnection());
serverConfig.setClientName(redssionProperties.getClientName());
serverConfig.setSubscriptionConnectionMinimumIdleSize(redssionProperties.getSubscriptionConnectionMinimumIdleSize());
serverConfig.setSubscriptionConnectionPoolSize(redssionProperties.getSubscriptionConnectionPoolSize());
serverConfig.setSlaveConnectionMinimumIdleSize(redssionProperties.getSlaveConnectionMinimumIdleSize());
serverConfig.setSlaveConnectionPoolSize(redssionProperties.getSlaveConnectionPoolSize());
serverConfig.setMasterConnectionMinimumIdleSize(redssionProperties.getMasterConnectionMinimumIdleSize());
serverConfig.setMasterConnectionPoolSize(redssionProperties.getMasterConnectionPoolSize());
serverConfig.setReadMode(redssionProperties.getReadMode());
serverConfig.setSubscriptionMode(redssionProperties.getSubscriptionMode());
serverConfig.setScanInterval(redssionProperties.getScanInterval());
serverConfig.setPassword(StringUtils.isNotBlank(passWord) && !"null".equals(passWord) ? passWord : null);
return Redisson.create(config);
}
}
3、业务代码中的应用。此处使用的是悲观锁，即必须拿到锁之后才能继续往下执行，也可使用乐观锁，tryLock，利用重试去获取锁
/**
* redissonClient
*/
@Resource
private RedissonClient redissonClient; /**
* 减库存
*
* @param trace 请求流水
* @param stockManageReq（stockId、decrNum）
* @return -1为失败，大于-1的正整数为减后的库存量，-2为库存不足无法减库存
*/
@Override
@ApiOperation(value = "减库存", notes = "减库存")
@RequestMapping(value = "/decrByStock", method = RequestMethod.POST, consumes = MediaType.APPLICATION_JSON_UTF8_VALUE, produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
public int decrByStock(@RequestHeader(name = "Trace") String trace, @RequestBody StockManageReq stockManageReq) { long startTime = System.currentTimeMillis(); LOGGER.reqPrint(Log.CACHE_SIGN, Log.CACHE_REQUEST, trace, "decrByStock", JSON.toJSONString(stockManageReq)); int res = 0;
String stockId = stockManageReq.getStockId();
Integer decrNum = stockManageReq.getDecrNum(); // 添加分布式锁
RLock stockLock = null; try {
if (null != stockId && null != decrNum) { stockId = PREFIX + stockId; // 添加分布式锁
stockLock = redissonClient.getFairLock(stockId); stockLock.lock(); // redis 减库存逻辑
String vStock = redisStockPool.get(stockId);
long realV = 0L;
if (StringUtils.isNotEmpty(vStock)) {
realV = Long.parseLong(vStock);
}
//库存数 大于等于 要减的数目，则执行减库存
if (realV >= decrNum) {
Long v = redisStockPool.decrBy(stockId, decrNum);
res = v.intValue();
} else {
res = -2;
} stockLock.unlock();
}
} catch (Exception e) {
LOGGER.error(trace, "decr sku stock failure.", e);
res = -1;
} finally {
if (stockLock != null && stockLock.isLocked() && stockLock.isHeldByCurrentThread()) {
stockLock.unlock();
}
LOGGER.respPrint(Log.CACHE_SIGN, Log.CACHE_RESPONSE, trace, "decrByStock", System.currentTimeMillis() - startTime, String.valueOf(res));
}
return res;
}
四、ab压测及结果分析
同样的，我们以5000的请求量100的并发量来压、tps在330左右，相对于zk做分布式锁来看，提升了10倍的性能，但仍然不能满足我们的要求
五、总结
redisson提供了丰富的分布式锁实现机制，并且使用起来相对比较简单方便，具体选用哪种锁，可以根据业务来选择，但在高并发的情况下，性能还是有些差强人意，下一篇，我们使用redis的watch来实现分布式锁。