redis分布式锁问题
1.如何避免死锁
在申请锁时，给这把锁设置一个过期时间
SET lock 1 EX 10 NX
2.锁超期问题
试想这样一种场景：
客户端 1 加锁成功，开始操作共享资源
客户端 1 操作共享资源的时间，「超过」了锁的过期时间，锁被「自动释放」
客户端 2 加锁成功，开始操作共享资源
客户端 1 操作共享资源完成，释放锁（但释放的是客户端 2 的锁）
这里存在两个严重的问题：
锁过期：客户端 1 操作共享资源耗时太久，导致锁被自动释放，之后被客户端 2 持有
释放别人的锁：客户端 1 操作共享资源完成后，却又释放了客户端 2 的锁
第一个问题，可能是我们评估操作共享资源的时间不准确导致的。可以通过延长过期时间来解决
第二个问题在于，一个客户端释放了其它客户端持有的锁。
解决办法是：客户端在加锁时，设置一个只有自己知道的「唯一标识」进去。例如，可以是自己的线程 ID，也可以是一个 UUID（随机且唯一）
SET lock $uuid EX 20 NX
3.锁释放的原子性问题
锁释放的伪代码
if redis.get("lock") == $uuid:
redis.del("lock")
客户端 1 执行 GET，判断锁是自己的
客户端 2 执行了 SET 命令，强制获取到锁（虽然发生概率比较低，但我们需要严谨地考虑锁的安全性模型）
客户端 1 执行 DEL，却释放了客户端 2 的锁
我们可以把这个逻辑，写成 Lua 脚本，让 Redis 来执行。因为 Redis 处理每一个请求是「单线程」执行的，在执行一个 Lua 脚本时，其它请求必须等待，直到这个 Lua 脚本处理完成
具体实现
1.基于 Spring Boot 引入redis
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- 对象池，使用redis时必须引入 -->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>
2.yaml配置
spring:
redis:
host: localhost
# 连接超时时间（记得添加单位，Duration）
timeout: 10000ms
# Redis默认情况下有16个分片，这里配置具体使用的分片
# database: 0
lettuce:
pool:
# 连接池最大连接数（使用负值表示没有限制） 默认 8
max-active: 8
# 连接池最大阻塞等待时间（使用负值表示没有限制） 默认 -1
max-wait: -1ms
# 连接池中的最大空闲连接 默认 8
max-idle: 8
# 连接池中的最小空闲连接 默认 0
min-idle: 0
3.
/**
* redis配置
* @author xiehengxing
* @date 2020/7/29 18:30
*/
@Configuration
@AutoConfigureAfter(RedisAutoConfiguration.class)
@EnableCaching
public class RedisConfig {
/**
* 默认情况下的模板只能支持RedisTemplate<String, String>，也就是只能存入字符串，因此支持序列化
*/
@Bean
public RedisTemplate<String, Serializable> redisCacheTemplate(LettuceConnectionFactory redisConnectionFactory) {
RedisTemplate<String, Serializable> template = new RedisTemplate<>();
template.setKeySerializer(new StringRedisSerializer());
template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
template.setConnectionFactory(redisConnectionFactory);
return template;
}
/**
* 配置使用注解的时候缓存配置，默认是序列化反序列化的形式，加上此配置则为 json 形式
*/
@Bean
public CacheManager cacheManager(RedisConnectionFactory factory) {
// 配置序列化
RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig();
RedisCacheConfiguration redisCacheConfiguration = config.serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer())).serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()));
return RedisCacheManager.builder(factory).cacheDefaults(redisCacheConfiguration).build();
}
}
/**
* 分布式共享锁
* @author xiehengxing
* @date 2020/8/2 10:40
*/
@Slf4j
@Component
public class RedisLock {
@Autowired
private RedisTemplate redisTemplate;
/**
* 共享锁默认时长（秒）
*/
private static final long TIMEOUT = 60;
/**
* 超时时长（秒）
*/
private static final long OVERTIME = 6;
/**
* 获取锁
* @param key
* @param requestId
* @return
*/
public boolean lock(String key, String requestId){
long startTime = System.currentTimeMillis();
for(;;) {
boolean locked = redisTemplate.opsForValue().setIfAbsent(key, requestId, TIMEOUT, TimeUnit.SECONDS);
if (locked) {
return true;
}
if ((System.currentTimeMillis() - startTime)/1000 > OVERTIME) {
return false;
}
try {
Thread.sleep(300);
} catch (InterruptedException e) {
log.error("线程被中断" + Thread.currentThread().getId(), e);
}
}
}
/**
* 使用lua脚本解锁
* @param key
* @param requestId
* @return
*/
public boolean unlock(String key, String requestId) {
if (StringUtils.isEmpty(key) || StringUtils.isEmpty(requestId)){
return false;
}
DefaultRedisScript<Long> redisScript = new DefaultRedisScript();
//用于解锁的lua脚本位置
redisScript.setScriptText(
"if redis.call('get', KEYS[1]) == ARGV[1] then " +
"return redis.call('del', KEYS[1]) " +
"else " +
"return 0" +
"end");
redisScript.setResultType(Long.class);
//没有指定序列化方式，默认使用上面配置的
Object result = redisTemplate.execute(redisScript, Collections.singletonList(key), requestId);
return result.equals(Long.valueOf(1));
}
}
另一种方案 Redisson
是否可以设计这样的方案：加锁时，先设置一个过期时间，然后我们开启一个「守护线程」，定时去检测这个锁的失效时间，如果锁快要过期了，操作共享资源还未完成，那么就自动对锁进行「续期」，重新设置过期时间。
Redisson 是一个 Java 语言实现的 Redis SDK 客户端，在使用分布式锁时，它就采用了「自动续期」的方案来避免锁过期，这个守护线程我们一般也把它叫做「看门狗」线程
除此之外，这个 SDK 还封装了很多易用的功能：
- 可重入锁
- 乐观锁
- 公平锁
- 读写锁
- Redlock
基于 ZooKeeper 的锁安全吗
基于 Redis 可以实现分布式锁，我们简单的讲解一下 ZooKeeper 怎么实现分布式锁。
客户端 1 和 2 都尝试创建「临时节点」，例如 /lock
假设客户端 1 先到达，则加锁成功，客户端 2 加锁失败
客户端 1 操作共享资源
客户端 1 删除 /lock 节点，释放锁
ZooKeeper 实现分布式锁的原理是 ZooKeeper 的节点不能重复创建，从而达到互斥的目的。
你应该也看到了，Zookeeper 不像 Redis 那样，需要考虑锁的过期时间问题，它是采用了「临时节点」，保证客户端 1 拿到锁后，只要连接不断，就可以一直持有锁。而且，如果客户端 1 异常崩溃了，那么这个临时节点会自动删除，保证了锁一定会被释放。
不错，没有锁过期的烦恼，还能在异常时自动释放锁，是不是觉得很完美？
其实不然。思考一下，客户端 1 创建临时节点后，Zookeeper 是如何保证让这个客户端一直持有锁呢？
原因就在于，**客户端 1 此时会与 Zookeeper 服务器维护一个 Session，这个 Session 会依赖客户端「定时心跳」来维持连接。**如果 Zookeeper 长时间收不到客户端的心跳，就认为这个 Session 过期了，也会把这个临时节点删除。
同样地，基于此问题，我们也讨论一下 GC 问题对 Zookeeper 的锁有何影响：
客户端 1 创建临时节点 /lock 成功，拿到了锁
客户端 1 发生长时间 GC
客户端 1 无法给 Zookeeper 发送心跳，Zookeeper 把临时节点「删除」
客户端 2 创建临时节点 /lock 成功，拿到了锁
客户端 1 GC 结束，它仍然认为自己持有锁（冲突）
可见，即使是使用 Zookeeper，也无法保证进程 GC、网络延迟异常场景下的安全性。所以，这里我们就能得出结论了：一个分布式锁，在极端情况下，不一定是安全的。
好，现在我们来总结一下 Zookeeper 在使用分布式锁时优劣：
Zookeeper 的优点：
不需要考虑锁的过期时间
watch 机制，加锁失败，可以 watch 等待锁释放，实现乐观锁
但它的劣势是：
性能不如 Redis
部署和运维成本高
客户端与 Zookeeper 的长时间失联，锁被释放问题
参考文章：https://blog.csdn.net/qq_39739458/article/details/107974476
https://blog.csdn.net/qq_46153765/article/details/121432399?spm=1001.2101.3001.6650.1&utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7ECTRLIST%7ERate-1-121432399-blog-107974476.pc_relevant_3mothn_strategy_recovery&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7ECTRLIST%7ERate-1-121432399-blog-107974476.pc_relevant_3mothn_strategy_recovery&utm_relevant_index=2
标签：return,过期,Zookeeper,redis,共享资源,分布式,客户端 From： https://www.cnblogs.com/l1pe1/p/16789493.html