多级缓存分析篇(二) 常用分布式锁分析

上篇主要讲了日常经常使用的哪些redis包及其配置的差异，这篇对基于redis的分布式锁，主要分享下自定义锁和redisson锁的使用和分析。

1.(自定义)Redis分布式锁

对于redis来说，非常适合做分布式锁来控制各个服务并发请求下造成的变量不一致、会串的问题。zookeeper锁其实也可以但不是很推荐，毕竟zk客户端是有某节点宕机自愈后、数据分发不同步的问题。

Redis锁的原理其实就是SETNX的原子性控制，建议一定要设定超时时间，防止资源竞争过于激烈导致线程长时间获取不到锁、发生timeout最终导致死锁。而像Mysql数据库采用的是InnoDB模式，默认参数innodb_lock_wait_timeout设置锁等待的时间是50s，一旦数据库锁超过这个时间就会报错。

SETNX的完整语法：SET key value [EX seconds|PX milliseconds] [NX|XX] [KEEPTTL]

必选参数说明：

SET：命令
key：待设置的key
value：设置的key的value，最好为随机字符串

可选参数说明：

NX：表示key不存在时才设置，如果存在则返回 null
XX：表示key存在时才设置，如果不存在则返回NULL
PX millseconds：设置过期时间，过期时间精确为毫秒
EX seconds：设置过期时间，过期时间精确为秒

这里提供一个自定义的Redis锁给大家参考使用：

package com.test.common.utils;import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisConnectionUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;import java.util.List;
import java.util.UUID;@Component
@Slf4j
public class RedisLockUtil {@Autowiredprivate RedisTemplate redisTemplate;/*** 默认锁超时时间（毫秒）*/private static final int DEFAULT_EXPIRE = 5*1000;/*** 锁名称前缀*/private static final String LOCK_PREFIX = "Coupon:";private RedisLock() {}/*** 获取锁* @param lockName    锁名称* @param acquireTimeout 等待获取锁的超时时间（毫秒）* @return  加锁成功后返回锁的唯一标识，未获取成功则返回null*/public String lock(String lockName, long acquireTimeout) {return lockWithTimeout(lockName, acquireTimeout, DEFAULT_EXPIRE);}/*** 获取锁* @param lockName   锁名称* @param acquireTimeout 等待获取锁的超时时间（毫秒）* @param timeout   锁超时时间，上锁后超过此时间则自动释放锁* @return   加锁成功后返回锁的唯一标识，未获取成功则返回null*/public String lockWithTimeout(String lockName, long acquireTimeout, long timeout){RedisConnectionFactory connectionFactory = redisTemplate.getConnectionFactory();RedisConnection redisConnection = connectionFactory.getConnection();/** 随机生成一个value */String identifier = UUID.randomUUID().toString();String lockKey = LOCK_PREFIX + lockName;int lockExpire = (int)(timeout / 1000);long end = System.currentTimeMillis() + acquireTimeout;  /** 获取锁的超时时间，超过这个时间则放弃获取锁 */while (System.currentTimeMillis() < end) {if (redisConnection.setNX(lockKey.getBytes(), identifier.getBytes())) {redisConnection.expire(lockKey.getBytes(), lockExpire);/** 获取锁成功，返回标识锁的value值，用于释放锁确认 */RedisConnectionUtils.releaseConnection(redisConnection, connectionFactory);return identifier;}/** 返回-1代表key没有设置超时时间，为key设置一个超时时间 */if (redisConnection.ttl(lockKey.getBytes()) == -1) {redisConnection.expire(lockKey.getBytes(), lockExpire);}try {Thread.sleep(10);} catch (InterruptedException e) {log.warn("获取分布式锁：线程中断！");Thread.currentThread().interrupt();}}RedisConnectionUtils.releaseConnection(redisConnection, connectionFactory);return null;}/*** 释放锁* @param lockName 锁名称* @param identifier 锁标识* @return*/public boolean releaseLock(String lockName, String identifier) {if (StringUtils.isEmpty(identifier)) return false;RedisConnectionFactory connectionFactory = redisTemplate.getConnectionFactory();RedisConnection redisConnection = connectionFactory.getConnection();String lockKey = LOCK_PREFIX + lockName;boolean releaseFlag = false;while (true) {try{/** 监视lock，准备开始事务 *///redisConnection.watch(lockKey.getBytes());byte[] valueBytes = redisConnection.get(lockKey.getBytes());/**  value为空表示锁不存在或已经被释放*/if(valueBytes == null){//redisConnection.unwatch();releaseFlag = false;break;}/** 通过前面返回的value值判断是不是该锁，若是该锁，则删除，释放锁 */String identifierValue = new String(valueBytes);if (identifier.equals(identifierValue)) {//redisConnection.multi();redisConnection.del(lockKey.getBytes());
//                  List<Object> results = redisConnection.exec();
//                  if (results == null) {
//                      continue;
//                  }releaseFlag = true;}
//              redisConnection.unwatch();break;}catch(Exception e){log.warn("释放锁异常", e);}}RedisConnectionUtils.releaseConnection(redisConnection, connectionFactory);return releaseFlag;}
}

使用Redis分布式锁用于领券场景的单元测试类：

package com.test.marketing.svc.client;import com.test.marketing.base.client.RedisLock;
import com.test.marketing.base.entity.CouponSummary;
import com.test.marketing.svc.init.MarketingServiceApplication;
import org.apache.commons.lang3.StringUtils;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import javax.annotation.Resource;
import java.util.stream.IntStream;@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = MarketingServiceApplication.class)
public class RedisLockTest {@Resourceprivate RedisLock redisLock;/***  测试并发领券／抢券场景下，使用redis分布锁累计领券总数*/@Testpublic void testCountReceive() {CouponSummary couponSummary = new CouponSummary();couponSummary.setReceivedQuantity(0l);IntStream.range(0,50).parallel().forEach( o->countReceiveQuantity(couponSummary) );log.info("Count Received quantities :" + couponSummary.getReceivedQuantity());Assert.assertNotEquals(50l, couponSummary.getReceivedQuantity().longValue());couponSummary.setReceivedQuantity(0l);IntStream.range(0,50).parallel().forEach( o->countReceiveQuantityWithLock("test",1000L, couponSummary) );log.info("Count Received quantities with lock :" + couponSummary.getReceivedQuantity());Assert.assertEquals(50l, couponSummary.getReceivedQuantity().longValue());}private void countReceiveQuantityWithLock(String lockName, Long acquireTimeout, CouponSummary couponSummary)  {String lockIdentify = redisLock.lock(lockName,acquireTimeout);if (StringUtils.isNotEmpty(lockIdentify)){countReceiveQuantity(couponSummary);redisLock.releaseLock(lockName, lockIdentify);}else{log.error("get lock failed.");}}private void countReceiveQuantity(CouponSummary couponSummary){couponSummary.setReceivedQuantity(couponSummary.getReceivedQuantity()+1);try {Thread.sleep(10l);} catch (InterruptedException e) {log.error("count thread was interrupted");}}}

2.Redisson分布式锁

Redisson是比较不错的开源框架，一般用于分布式锁或注册中心控制。Redisson实现的是可重入锁，接口RLock通过RedissonLock类来实现主要逻辑的，

2.1 使用案例

Redisson的使用方式其实很简单：

    @Resourceprivate RedissonClient redissonClient;//如下为代码片段：//若不加锁则会产生超卖RLock lock = redissonClient.getLock("stock:" + productId);try {lock.lock(10, TimeUnit.SECONDS);int stockAmount = stockService.get(productId).getStockAmount();log.info("剩余库存量为：{}", stockAmount);if (stockAmount <= 0) {return false;}String orderNo = UUID.randomUUID().toString().replace(".", "").toUpperCase();/*原子性操作(例如减库存操作等)...*/} catch (Exception ex) {log.error("下单失败", ex);} finally {lock.unlock();}

备注：这里的getLock()默认是指的非公平锁，redisson也提过了公平锁getFairLock和多锁getMultiLock():RedissonMultiLock，也支持异步方式lock.tryLockAsync(100,10,TimeUnit.SECONDS):Future<T>，很方便使用。

其它细节更多可参考：Redisson基本用法 - 废物大师兄 - 博客园

2.2 配置Redisson

Redisson推荐redis集群节点至少为3个(奇数个)，redis cluster集群版本个人推荐在5.0.3以上。

YAML配置方式(redisson.yml)：

clusterServersConfig:idleConnectionTimeout: 10000pingTimeout: 1000connectTimeout: 10000timeout: 3000retryAttempts: 3retryInterval: 1500reconnectionTimeout: 3000failedAttempts: 3password: nullsubscriptionsPerConnection: 5clientName: nullloadBalancer: !<org.redisson.connection.balancer.RoundRobinLoadBalancer> {}subscriptionConnectionMinimumIdleSize: 1subscriptionConnectionPoolSize: 50slaveConnectionMinimumIdleSize: 5slaveConnectionPoolSize: 128masterConnectionMinimumIdleSize: 5masterConnectionPoolSize: 128readMode: "SLAVE"subscriptionMode: "SLAVE"nodeAddresses:- "redis://172.17.17.5:30004"- "redis://172.17.17.5:30005"- "redis://172.17.17.5:30006"scanInterval: 1000
threads: 0
nettyThreads: 0
codec: !<org.redisson.codec.SnappyCodec> {}

配置类方式：

/*** redisson配置类*/
@Configuration
public class RedissonConfig {@Beanpublic RedissonClient redissonClient() {Config config = new Config();config.useClusterServers().setScanInterval(2000).addNodeAddress("redis://10.102.5.11:6379","redis://10.102.5.11:6380").addNodeAddress("redis://redis://10.102.5.11:6381");RedissonClient redisson = Redisson.create(config);return redisson;}}

好了，基于redis的分布式锁就讨论到这里，下一讲还是回归大本营，谈谈Spring官方对Cache的顶层设计及其Redis的桥接是如何实现的多级缓存分析篇(三) Spring本地缓存源码分析，欢迎来交流留言哦~~