“分布式锁”是用来解决分布式应用中“并发冲突”的一种常用手段,实现方式一般有基于zookeeper及基于redis二种。具体到业务场景中,我们要考虑二种情况:
一、抢不到锁的请求,允许丢弃(即:忽略)
二、并发请求,不论哪一条都必须要处理的场景(即:不允许丢数据)
比如:一个订单,客户正在前台修改地址,管理员在后台同时修改备注。地址和备注字段的修改,都必须正确更新,这二个请求同时到达的话,如果不借助db的事务,很容易造成行锁竞争,但用事务的话,db的性能显然比不上redis轻量。
解决思路:A,B二个请求,谁先抢到分布式锁(假设A先抢到锁),谁先处理,抢不到的那个(即:B),在一旁不停等待重试,重试期间一旦发现获取锁成功,即表示A已经处理完,把锁释放了。这时B就可以继续处理了。
但有二点要注意:
a、需要设置等待重试的最长时间,否则如果A处理过程中有bug,一直卡死,或者未能正确释放锁,B就一直会等待重试,但是又永远拿不到锁。
b、等待最长时间,必须小于锁的过期时间。否则,假设锁2秒过期自动释放,但是A还没处理完(即:A的处理时间大于2秒),这时锁会因为redis key过期“提前”误释放,B重试时拿到锁,造成A,B同时处理。(注:可能有同学会说,不设置锁的过期时间,不就完了么?理论上讲,确实可以这么做,但是如果业务代码有bug,导致处理完后没有unlock,或者根本忘记了unlock,分布式锁就会一直无法释放。所以综合考虑,给分布式锁加一个“保底”的过期时间,让其始终有机会自动释放,更为靠谱)
package com.cnblogs.yjmyzz.redisdistributionlock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.util.StringUtils; import java.util.UUID; import java.util.concurrent.TimeUnit; / * 利用redis获取分布式锁 * * @author 菩提树下的杨过 * @blog http://yjmyzz.cnblogs.com/ */ public class RedisLock {
private StringRedisTemplate redisTemplate; private Logger logger = LoggerFactory.getLogger(this.getClass()); / * simple lock尝试获取锅的次数 */ private int retryCount = 3; / * 每次尝试获取锁的重试间隔毫秒数 */ private int waitIntervalInMS = 100; public RedisLock(StringRedisTemplate redisTemplate) {
this.redisTemplate = redisTemplate; } / * 利用redis获取分布式锁(未获取锁的请求,允许丢弃!) * * @param redisKey 锁的key值 * @param expireInSecond 锁的自动释放时间(秒) * @return * @throws DistributionLockException */ public String simpleLock(final String redisKey, final int expireInSecond) throws DistributionLockException {
String lockValue = UUID.randomUUID().toString(); boolean flag = false; if (StringUtils.isEmpty(redisKey)) {
throw new DistributionLockException("key is empty!"); } if (expireInSecond <= 0) {
throw new DistributionLockException("expireInSecond must be bigger than 0"); } try {
for (int i = 0; i < retryCount; i++) {
boolean success = redisTemplate.opsForValue().setIfAbsent(redisKey, lockValue, expireInSecond, TimeUnit.SECONDS); if (success) {
flag = true; break; } try {
TimeUnit.MILLISECONDS.sleep(waitIntervalInMS); } catch (Exception ignore) {
logger.warn("redis lock fail: " + ignore.getMessage()); } } if (!flag) {
throw new DistributionLockException(Thread.currentThread().getName() + " cannot acquire lock now ..."); } return lockValue; } catch (DistributionLockException be) {
throw be; } catch (Exception e) {
logger.warn("get redis lock error, exception: " + e.getMessage()); throw e; } } / * 利用redis获取分布式锁(未获取锁的请求,将在timeoutSecond时间范围内,一直等待重试) * * @param redisKey 锁的key值 * @param expireInSecond 锁的自动释放时间(秒) * @param timeoutSecond 未获取到锁的请求,尝试重试的最久等待时间(秒) * @return * @throws DistributionLockException */ public String lock(final String redisKey, final int expireInSecond, final int timeoutSecond) throws DistributionLockException {
String lockValue = UUID.randomUUID().toString(); boolean flag = false; if (StringUtils.isEmpty(redisKey)) {
throw new DistributionLockException("key is empty!"); } if (expireInSecond <= 0) {
throw new DistributionLockException("expireInSecond must be greater than 0"); } if (timeoutSecond <= 0) {
throw new DistributionLockException("timeoutSecond must be greater than 0"); } if (timeoutSecond >= expireInSecond) {
throw new DistributionLockException("timeoutSecond must be less than expireInSecond"); } try {
long timeoutAt = System.currentTimeMillis() + timeoutSecond * 1000; while (true) {
boolean success = redisTemplate.opsForValue().setIfAbsent(redisKey, lockValue, expireInSecond, TimeUnit.SECONDS); if (success) {
flag = true; break; } if (System.currentTimeMillis() >= timeoutAt) {
break; } try {
TimeUnit.MILLISECONDS.sleep(waitIntervalInMS); } catch (Exception ignore) {
logger.warn("redis lock fail: " + ignore.getMessage()); } } if (!flag) {
throw new DistributionLockException(Thread.currentThread().getName() + " cannot acquire lock now ..."); } return lockValue; } catch (DistributionLockException be) {
throw be; } catch (Exception e) {
logger.warn("get redis lock error, exception: " + e.getMessage()); throw e; } } / * 锁释放 * * @param redisKey * @param lockValue */ public void unlock(final String redisKey, final String lockValue) {
if (StringUtils.isEmpty(redisKey)) {
return; } if (StringUtils.isEmpty(lockValue)) {
return; } try {
String currLockVal = redisTemplate.opsForValue().get(redisKey); if (currLockVal != null && currLockVal.equals(lockValue)) {
boolean result = redisTemplate.delete(redisKey); if (!result) {
logger.warn(Thread.currentThread().getName() + " unlock redis lock fail"); } else {
logger.info(Thread.currentThread().getName() + " unlock redis lock:" + redisKey + " successfully!"); } } } catch (Exception je) {
logger.warn(Thread.currentThread().getName() + " unlock redis lock error:" + je.getMessage()); } } }
然后写个spring-boot来测试一下:
package com.cnblogs.yjmyzz.redisdistributionlock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.data.redis.core.StringRedisTemplate; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @SpringBootApplication public class RedisDistributionLockApplication {
private static Logger logger = LoggerFactory.getLogger(RedisDistributionLockApplication.class); public static void main(String[] args) throws InterruptedException {
ConfigurableApplicationContext applicationContext = SpringApplication.run(RedisDistributionLockApplication.class, args); //初始化 StringRedisTemplate redisTemplate = applicationContext.getBean(StringRedisTemplate.class); RedisLock redisLock = new RedisLock(redisTemplate); String lockKey = "lock:test"; CountDownLatch start = new CountDownLatch(1); CountDownLatch threadsLatch = new CountDownLatch(2); final int lockExpireSecond = 5; final int timeoutSecond = 3; Runnable lockRunnable = () -> {
String lockValue = ""; try {
//等待发令枪响,防止线程抢跑 start.await(); //允许丢数据的简单锁示例 lockValue = redisLock.simpleLock(lockKey, lockExpireSecond); //不允许丢数据的分布式锁示例 //lockValue = redisLock.lock(lockKey, lockExpireSecond, timeoutSecond); //停一会儿,故意让后面的线程抢不到锁 TimeUnit.SECONDS.sleep(2); logger.info(String.format("%s get lock successfully, value:%s", Thread.currentThread().getName(), lockValue)); } catch (Exception e) {
e.printStackTrace(); } finally {
redisLock.unlock(lockKey, lockValue); //执行完后,计数减1 threadsLatch.countDown(); } }; Thread t1 = new Thread(lockRunnable, "T1"); Thread t2 = new Thread(lockRunnable, "T2"); t1.start(); t2.start(); //预备:开始! start.countDown(); //等待所有线程跑完 threadsLatch.await(); logger.info("======>done!!!"); } }
发布者:全栈程序员-站长,转载请注明出处:https://javaforall.net/205379.html原文链接:https://javaforall.net
