redis分布式锁两种应用场景

“分布式锁”是用来解决分布式应用中“并发冲突”的一种常用手段，实现方式一般有基于zookeeper及基于redis二种。具体到业务场景中，我们要考虑二种情况：一、抢不到锁的请求，允许丢弃(即：忽略）比如：一些不是很重要的场景，比如“监控数据持续上报”，某一篇文章的“已读/未读”标识位更新，对于同一个id，如果并发的请求同时到达，只要有一个请求处理成功，就算成功。用活动图表示如下：二、并发请求，不论

xiaolege_

9482人浏览 · 2020-11-19 10:40:37

xiaolege_ · 2020-11-19 10:40:37 发布

“分布式锁”是用来解决分布式应用中“并发冲突”的一种常用手段，实现方式一般有基于zookeeper及基于redis二种。

这里我们分析下基于redis得场景和实现。

单节点部署场景

举例说明，系统A和系统B是两个部署在不同节点的相同应用（集群部署），这时客户端请求传来，两个系统都受到了请求，并且该请求是对数据表进行插入操作，如果这个时候不加锁来控制，可能会导致数据库新增两条记录，这时系统也不能允许的，由于是在不同应用内，在单个应用内加JVM级别的锁，另一个应用是感知不到的，这时需要用到分布式锁。
接下来我们看看这种场景如何实现安全的分布式锁，由于是单节点部署场景，我们可以用setnx命令，以请求的唯一主键作为key，由于该操作是原子操作，当系统A设值成功后，系统B是无法设置成功的，这时A就可以进行查询并插入操作，操作数据库完成后，删除key，此时系统B才能设值成功，但是由于查询到数据库有记录，所以并不会插入数据，这样就解决了该问题。但是这里会有个问题，如果redis挂机了，这里的锁不是永远都不释放了吗，所以为了解决这个问题，redis提供了set命令，可传入超时时间的，那么在指定的时间范围内，如果没有释放锁，则该锁自动过期。如果执行时间超过超时时间呢，比如系统A还未执行完任务，就释放了锁，系统B接着执行任务，这时，系统A执行完了，把锁删掉（此时删除的时系统B获取的锁）。
- 方案一：为了避免这种情况，在del锁之前可以做一个判断，验证key对应的value是不是自己线程的ID.如果要考虑原子性问题，可以使用Lua脚本来实现，保证验证和删除的原子性。
- 方案二：我们可以让获得锁的线程开启一个守护线程，用来给快要过期的锁加长超时时间。当系统A中的线程执行完任务，再显式关掉守护线程。

具体到业务场景中，我们要考虑二种情况：

一、抢不到锁的请求，允许丢弃(即：忽略）

比如：一些不是很重要的场景，比如“监控数据持续上报”，某一篇文章的“已读/未读”标识位更新，对于同一个id，如果并发的请求同时到达，只要有一个请求处理成功，就算成功。

用活动图表示如下：

点击查看原图

二、并发请求，不论哪一条都必须要处理的场景(即：不允许丢数据)

比如：一个订单，客户正在前台修改地址，管理员在后台同时修改备注。地址和备注字段的修改，都必须正确更新，这二个请求同时到达的话，如果不借助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，分布式锁就会一直无法释放。所以综合考虑，给分布式锁加一个“保底”的过期时间，让其始终有机会自动释放，更为靠谱）

用活动图表示如下：

点击查看原图

写了一个简单的工具类：

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

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!!!");

}

　用2个线程模拟并发场景，跑起来后，输出如下：

点击查看原图

可以看到T2线程没抢到锁，直接抛出了预期的异常。

把44行的注释打开，即：换成不允许丢数据的模式，再跑一下：

点击查看原图

可以看到，T1先抢到锁，然后经过2秒的处理后，锁释放，这时T2重试拿到了锁，继续处理，最终释放。

文章参考：

基于redis的分布式锁二种应用场景

https://www.cnblogs.com/yjmyzz/

向您推荐>>Eolink开发者社区

权威｜前沿｜技术｜干货｜国内首个API全生命周期开发者社区

更多推荐

深入理解 Mocha 测试框架：从零实现一个 Mocha

前言什么是自动化测试自动化测试在很多团队中都是Devops环节中很难执行起来的一个环节，主要原因在于测试代码的编写工作很难抽象，99%的场景都需要和业务强绑定，而且写测试代码的编写工作量往往比编写实际业务代码的工作量更多。在一些很多业务场景中投入产出比很低，适合写自动化测试的应该是那些中长期业务以及一些诸如组件一样的基础库。自动化测试是个比较大的概念，其中分类也比较多，比如单元测试，端对端测试，集

云原生

ELK实现containerd的容器日志采集展示【基于logging的全栈监测】

企业级ELK Stack构建介绍

云原生

(20200916 Solved)docker-compose up创建容器自动退出

问题描述如题，创建容器后自动退出了。并且docker start container无效解决方案原因是缺失了控制终端的配置，需要在docker-compose.yml中增加tty:true ，有时候这样也不行，需要再增加一个command:/bin/bash，命令不一定是这个，需要是一个不会退出的命令，然后用-d后台启动容器。Referencesdocker-compose启动容器后自动退出...