"文章来源：https://www.cnblogs.com/niudaxianren/p/14754597.html redis 分布式锁 -java 实现 1、为什么要使用分布式锁如果在一个分布式系统中，我们从数据库中读取一个数据，然后修改保存，这种情况很容易遇到并发问题。因为读取和更新保存不是一 ...."

someone1
AI专业者开发社区 1 号成员
AI开发者社区 • 0 回帖 • 0 浏览 • 5 个月前

redis 解决并发 java

文章来源：https://www.cnblogs.com/niudaxianren/p/14754597.html

redis 分布式锁 -java 实现

1、为什么要使用分布式锁

如果在一个分布式系统中，我们从数据库中读取一个数据，然后修改保存，这种情况很容易遇到并发问题。因为读取和更新保存不是一个原子操作，在并发时就会导致数据的不正确。这种场景其实并不少见，比如电商秒杀活动，库存数量的更新就会遇到。如果是单机应用，直接使用本地锁就可以避免。如果是分布式应用，本地锁派不上用场，这时就需要引入分布式锁来解决。

由此可见分布式锁的目的其实很简单，就是为了保证多台服务器在执行某一段代码时保证只有一台服务器执行。

2、为了保证分布式锁的可用性，至少要确保锁的实现要同时满足以下几点：

互斥性。在任何时刻，保证只有一个客户端持有锁。
不能出现死锁。如果在一个客户端持有锁的期间，这个客户端崩溃了，也要保证后续的其他客户端可以上锁。
保证上锁和解锁都是同一个客户端。

3、一般来说，实现分布式锁的方式有以下几种：

使用 MySQL，基于唯一索引。
使用 ZooKeeper，基于临时有序节点。
使用 Redis，基于 set 命令（2.6.12 版本开始）。

本篇文章主要讲解 Redis 的实现方式。

4、用到的 redis 命令

锁的实现主要基于 redis 的 SET 命令（SET 详细解释参考这里），我们来看 SET 的解释：

SET key value [EX seconds] [PX milliseconds] [NX|XX]

将字符串值 value 关联到 key 。
如果 key 已经持有其他值， SET 就覆写旧值，无视类型。
对于某个原本带有生存时间（TTL）的键来说，当 SET 命令成功在这个键上执行时，这个键原有的 TTL 将被清除。
可选参数

从 Redis 2.6.12 版本开始， SET 命令的行为可以通过一系列参数来修改：

EX second ：设置键的过期时间为 second 秒。 SET key value EX second 效果等同于 SETEX key second value 。
PX millisecond ：设置键的过期时间为 millisecond 毫秒。 SET key value PX millisecond 效果等同于 PSETEX key millisecond value 。
NX ：只在键不存在时，才对键进行设置操作。 SET key value NX 效果等同于 SETNX key value 。
XX ：只在键已经存在时，才对键进行设置操作。

加锁：使用SET key value [PX milliseconds] [NX]命令，如果 key 不存在，设置 value，并设置过期时间 (加锁成功)。如果已经存在 lock(也就是有客户端持有锁了)，则设置失败 (加锁失败)。

解锁：使用 del 命令，通过删除键值释放锁。释放锁之后，其他客户端可以通过 set 命令进行加锁。

5、上面第二项，说了分布式锁，要考虑的问题，下面讲解一下

5.1、互斥性。在任何时刻，保证只有一个客户端持有锁

redis 命令是原子性的，只要客户端调用 redis 的命令 SET key value [PX milliseconds] [NX] 执行成功，就算加锁成功了

5.2、不能出现死锁。如果在一个客户端持有锁的期间，这个客户端崩溃了，也要保证后续的其他客户端可以上锁。

set 命令 px 设置了过期时间，key 过期失效了，就能避免死锁了

5.3 保证上锁和解锁都是同一个客户端。

释放锁 (删除 key) 的时候，只要确保是当前客户端设置的 value 才去删除 key 即可，采用 lua 脚本来实现

在 Redis 中，执行 Lua 语言是原子性，也就是说 Redis 执行 Lua 的时候是不会被中断的，具备原子性，这个特性有助于 Redis 对并发数据一致性的支持。

6、java 代码实现

先把需要的 jar 包引入

        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.3</version>
        </dependency>

加锁设置参数的实体类

import lombok.Data;
// 加锁设置的参数
@Data
public class LockParam {
// 锁的 key
private String lockKey;
// 尝试获得锁的时间（单位：毫秒），默认值：3000 毫秒
private Long tryLockTime;
// 尝试获得锁后，持有锁的时间（单位：毫秒），默认值：5000 毫秒
private Long holdLockTime;
<span class="hljs-keyword">public</span> LockParam(String lockKey){
    <span class="hljs-keyword">this</span>(lockKey,<span class="hljs-number">1000</span>*<span class="hljs-number">3L</span>,<span class="hljs-number">1000</span>*<span class="hljs-number">5L</span>);
};
<span class="hljs-keyword">public</span> LockParam(String lockKey,<span class="hljs-built_in">Long</span> tryLockTime){
    <span class="hljs-keyword">this</span>(lockKey,tryLockTime,<span class="hljs-number">1000</span>*<span class="hljs-number">5L</span>);
};
<span class="hljs-keyword">public</span> LockParam(String lockKey,<span class="hljs-built_in">Long</span> tryLockTime,<span class="hljs-built_in">Long</span> holdLockTime){
    <span class="hljs-keyword">this</span>.lockKey = lockKey;
    <span class="hljs-keyword">this</span>.tryLockTime = tryLockTime;
    <span class="hljs-keyword">this</span>.holdLockTime = holdLockTime;
};

}

redis 分布式具体代码实现

import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;
import java.util.Collections;
import java.util.UUID;
/**


redis 分布式锁
*/
@Slf4j
public class RedisLock {
// 锁 key 的前缀
private final static String prefix_key = "redisLock:";
// 释放锁的 lua 脚本
private final static  String unLockScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
// 执行 unLockScript 脚本，释放锁成功值
private final static  Long unLockSuccess = 1L;
// 加锁设置的参数（key 值、超时时间、持有锁的时间）
private LockParam lockParam;
// 尝试获得锁的截止时间【lockParam.getTryLockTime()+System.currentTimeMillis()】
private Long tryLockEndTime;
//redis 加锁的 key
private String redisLockKey;
//redis 加锁的 vlaus
private String redisLockValue;
//redis 加锁的成功标示
private Boolean holdLockSuccess= Boolean.FALSE;
//jedis 实例
private Jedis jedis;
// 获取 jedis 实例
private Jedis getJedis(){
return this.jedis;
}
// 关闭 jedis
private void closeJedis(Jedis jedis){
jedis.close();
jedis = null;
}
public RedisLock(LockParam lockParam){
if(lockParam==null){
new RuntimeException("lockParam is null");
}
if(lockParam.getLockKey()==null || lockParam.getLockKey().trim().length()==0){
new RuntimeException("lockParam lockKey is error");
}
this.lockParam = lockParam;
 <span class="hljs-variable language_">this</span>.<span class="hljs-property">tryLockEndTime</span> = lockParam.<span class="hljs-title function_">getTryLockTime</span>()+<span class="hljs-title class_">System</span>.<span class="hljs-title function_">currentTimeMillis</span>();
 <span class="hljs-variable language_">this</span>.<span class="hljs-property">redisLockKey</span> = prefix_key.<span class="hljs-title function_">concat</span>(lockParam.<span class="hljs-title function_">getLockKey</span>());
 <span class="hljs-variable language_">this</span>.<span class="hljs-property">redisLockValue</span> = <span class="hljs-variable constant_">UUID</span>.<span class="hljs-title function_">randomUUID</span>().<span class="hljs-title function_">toString</span>().<span class="hljs-title function_">replaceAll</span>(<span class="hljs-string">"-"</span>,<span class="hljs-string">""</span>);

 <span class="hljs-comment">//todo 到时候可以更换获取Jedis实例的实现</span>
 jedis = <span class="hljs-keyword">new</span> <span class="hljs-title class_">Jedis</span>(<span class="hljs-string">"127.0.0.1"</span>,<span class="hljs-number">6379</span>);

}
/**


加锁


@return 成功返回 true，失败返回 false
*/
public boolean lock() {
while(true){
// 判断是否超过了，尝试获取锁的时间
if(System.currentTimeMillis()>tryLockEndTime){
return false;
}
// 尝试获取锁
holdLockSuccess = tryLock();
if(Boolean.TRUE.equals(holdLockSuccess)){
return true;// 获取锁成功
}
 <span class="hljs-keyword">try</span> {
     <span class="hljs-comment">//获得锁失败，休眠50毫秒再去尝试获得锁,避免一直请求redis，导致redis cpu飙升</span>
     <span class="hljs-title class_">Thread</span>.<span class="hljs-title function_">sleep</span>(<span class="hljs-number">50</span>);
 } <span class="hljs-keyword">catch</span> (<span class="hljs-title class_">InterruptedException</span> e) {
     e.<span class="hljs-title function_">printStackTrace</span>();
 }

}
}


/**

执行一次加锁操作：成功返回 true 失败返回 false
@return 成功返回 true，失败返回 false
*/
private boolean tryLock() {
try {
String result = getJedis().set(redisLockKey,redisLockValue, "NX", "PX", lockParam.getHoldLockTime());
if ("OK".equals(result)) {
return true;
}
}catch (Exception e){
log.warn("tryLock failure redisLockKey:{} redisLockValue:{} lockParam:{}",redisLockKey,redisLockValue,lockParam,e);
}
return false;
}

/**

解锁
@return 成功返回true，失败返回false
*/
public Boolean unlock() {
Object result = null;
try {
// 获得锁成功，才执行 lua 脚本
if(Boolean.TRUE.equals(holdLockSuccess)){
// 执行 Lua 脚本
result = getJedis().eval(unLockScript, Collections.singletonList(redisLockKey), Collections.singletonList(redisLockValue));
if (unLockSuccess.equals(result)) {// 释放成功
return true;
}
}
} catch (Exception e) {
log.warn("unlock failure redisLockKey:{} redisLockValue:{} lockParam:{} result:{}",redisLockKey,redisLockValue,lockParam,result,e);
} finally {
this.closeJedis(jedis);
}
return false;
}
}

redis 分布式锁使用

import lombok.extern.slf4j.Slf4j;
@Slf4j
public class test {
static String lockKey = "666";
public static void main(String[] args) throws InterruptedException {
log.info("下面测试两个线程同时，抢占锁的结果");
Thread thread1 = new Thread(()->{
testRedisLock();
});
thread1.setName("我是线程 1");
Thread thread2 = new Thread(()->{
testRedisLock();
});
thread2.setName("我是线程 2");
    <span class="hljs-comment">//同时启动线程</span>
    thread1.start();
    thread2.start();

    Thread.sleep(<span class="hljs-number">1000</span>*<span class="hljs-number">20</span>);
    log.info(<span class="hljs-string">"-----------------我是一条分割线----------------"</span>);
    log.info(<span class="hljs-string">""</span>);
    log.info(<span class="hljs-string">""</span>);
    log.info(<span class="hljs-string">""</span>);


    log.info(<span class="hljs-string">"下面是测试  一个线程获取锁成功后，由于业务执行时间超过了设置持有锁的时间，是否会把其他线程持有的锁给释放掉"</span>);
    <span class="hljs-type">Thread</span> <span class="hljs-variable">thread3</span> <span class="hljs-operator">=</span> <span class="hljs-keyword">new</span> <span class="hljs-title class_">Thread</span>(()-&gt;{
        testRedisLock2();
    });
    thread3.setName(<span class="hljs-string">"我是线程3"</span>);
    thread3.start();

    Thread.sleep(<span class="hljs-number">1000</span>*<span class="hljs-number">1</span>);<span class="hljs-comment">//暂停一秒是为了让线程3获的到锁</span>
    <span class="hljs-type">Thread</span> <span class="hljs-variable">thread4</span> <span class="hljs-operator">=</span> <span class="hljs-keyword">new</span> <span class="hljs-title class_">Thread</span>(()-&gt;{
        testRedisLock();
    });
    thread4.setName(<span class="hljs-string">"我是线程4"</span>);
    thread4.start();
}

<span class="hljs-keyword">public</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> <span class="hljs-title function_">testRedisLock</span><span class="hljs-params">()</span>{
    <span class="hljs-type">LockParam</span> <span class="hljs-variable">lockParam</span> <span class="hljs-operator">=</span> <span class="hljs-keyword">new</span> <span class="hljs-title class_">LockParam</span>(lockKey);
    lockParam.setTryLockTime(<span class="hljs-number">2000L</span>);<span class="hljs-comment">//2秒时间尝试获得锁</span>
    lockParam.setHoldLockTime(<span class="hljs-number">1000</span>*<span class="hljs-number">10L</span>);<span class="hljs-comment">//获得锁成功后持有锁10秒时间</span>
    <span class="hljs-type">RedisLock</span> <span class="hljs-variable">redisLock</span> <span class="hljs-operator">=</span> <span class="hljs-keyword">new</span> <span class="hljs-title class_">RedisLock</span>(lockParam);
    <span class="hljs-keyword">try</span> {
        <span class="hljs-type">Boolean</span> <span class="hljs-variable">lockFlag</span> <span class="hljs-operator">=</span> redisLock.lock();
        log.info(<span class="hljs-string">"加锁结果：{}"</span>,lockFlag);
        <span class="hljs-keyword">if</span>(lockFlag){
            <span class="hljs-keyword">try</span> {
                <span class="hljs-comment">//20秒模拟处理业务代码时间</span>
                Thread.sleep(<span class="hljs-number">1000</span>*<span class="hljs-number">5L</span>);
            } <span class="hljs-keyword">catch</span> (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }<span class="hljs-keyword">catch</span> (Exception e) {
        log.info(<span class="hljs-string">"testRedisLock e----&gt;"</span>,e);
    }<span class="hljs-keyword">finally</span> {
        <span class="hljs-type">boolean</span> <span class="hljs-variable">unlockResp</span> <span class="hljs-operator">=</span> redisLock.unlock();
        log.info(<span class="hljs-string">"释放锁结果:{}"</span>,unlockResp);
    }
}


<span class="hljs-keyword">public</span> <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> <span class="hljs-title function_">testRedisLock2</span><span class="hljs-params">()</span>{
    <span class="hljs-type">LockParam</span> <span class="hljs-variable">lockParam</span> <span class="hljs-operator">=</span> <span class="hljs-keyword">new</span> <span class="hljs-title class_">LockParam</span>(lockKey);
    lockParam.setTryLockTime(<span class="hljs-number">1000</span>*<span class="hljs-number">2L</span>);<span class="hljs-comment">//2秒时间尝试获得锁</span>
    lockParam.setHoldLockTime(<span class="hljs-number">1000</span>*<span class="hljs-number">2L</span>);<span class="hljs-comment">//获得锁成功后持有锁2秒时间</span>
    <span class="hljs-type">RedisLock</span> <span class="hljs-variable">redisLock</span> <span class="hljs-operator">=</span> <span class="hljs-keyword">new</span> <span class="hljs-title class_">RedisLock</span>(lockParam);
    <span class="hljs-keyword">try</span> {
        <span class="hljs-type">Boolean</span> <span class="hljs-variable">lockFlag</span> <span class="hljs-operator">=</span> redisLock.lock();
        log.info(<span class="hljs-string">"加锁结果：{}"</span>,lockFlag);
        <span class="hljs-keyword">if</span>(lockFlag){
            <span class="hljs-keyword">try</span> {
                <span class="hljs-comment">//10秒模拟处理业务代码时间</span>
                Thread.sleep(<span class="hljs-number">1000</span>*<span class="hljs-number">10L</span>);
            } <span class="hljs-keyword">catch</span> (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }<span class="hljs-keyword">catch</span> (Exception e) {
        log.info(<span class="hljs-string">"testRedisLock e----&gt;"</span>,e);
    }<span class="hljs-keyword">finally</span> {
        <span class="hljs-type">boolean</span> <span class="hljs-variable">unlockResp</span> <span class="hljs-operator">=</span> redisLock.unlock();
        log.info(<span class="hljs-string">"释放锁结果:{}"</span>,unlockResp);
    }
}

}

这是代码在执行过程中，通过 redis 可视化工具看到的效果，可以参考一下～

控制台日志打印结果

15:02:28.569 [main] INFO com.test.test - 下面测试两个线程同时，抢占锁的结果
15:02:28.645 [我是线程2] INFO com.test.test - 加锁结果：true
15:02:30.618 [我是线程1] INFO com.test.test - 加锁结果：false
15:02:30.620 [我是线程1] INFO com.test.test - 释放锁结果:false
15:02:33.652 [我是线程2] INFO com.test.test - 释放锁结果:true
15:02:48.614 [main] INFO com.test.test - -----------------我是一条分割线----------------
15:02:48.614 [main] INFO com.test.test - 
15:02:48.614 [main] INFO com.test.test - 
15:02:48.614 [main] INFO com.test.test - 
15:02:48.614 [main] INFO com.test.test - 下面是测试  一个线程获取锁成功后，由于业务执行时间超过了设置持有锁的时间，是否会把其他线程持有的锁给释放掉
15:02:48.616 [我是线程3] INFO com.test.test - 加锁结果：true
15:02:50.645 [我是线程4] INFO com.test.test - 加锁结果：true
15:02:55.647 [我是线程4] INFO com.test.test - 释放锁结果:true
15:02:58.621 [我是线程3] INFO com.test.test - 释放锁结果:false

可以看到多个线程竞争一把锁的时候，保证了只有一个线程持有锁
分割线下面的日志也能看出，一个线程持有了锁，由于处理业务代码时间，超过了设置持有锁的时间，通过 lua 脚本释放锁的时候，也不会把其他线程持有的锁给释放掉，保证了安全释放了锁

7、分布式锁实际使用中需要注意的一些问题

假设有这样一个场景：有一个修改订单状态的接口，订单状态修改为失败，就不允许在修改为其他状态了；
在单台机器上，在代码方法上加了 synchronized 来做并发控制，由于代码逻辑比较复杂，现在它的 TPS 是 1，一秒就只能处理一个订单。
后面对这个系统做集群，部署了一百台, 那么这个接口性能就提升了 100 倍了。
但是 synchronized 是进程级别的锁，在集群环境下 synchronized 没办法控制其他服务器下线程并发访问临界代码了，后面就采用了分布式锁来做并发控制。

7.1、那么使用分布锁要注意什么了？

7.1.1、锁粒度

如果分布式锁的 key 设置的是 redisLock:updateOrderStatus 相当于集群下对这个接口加了相同的一把大锁，按照上面那个场景 TPS 就变成 1 了，集群部署就浪费了。

7.1.2、那么如何控制锁粒度了？

平常我们修改订单的时候都有订单号，那么分布式的 key 可以设置为：redisLock:updateOrderStatus:{orderCode} ，{orderCode} 执行的时候动态的替换成订单编号，那么锁粒度就控制到这条订单了，就跟数据库从表锁变成了行锁一样，接口支持更高的并发了。

7.1.3、获取锁时间

如果时间设置的太长：用户就会等待太久才能得到响应结果
太短：频繁获取锁失败，用户体验性也不好
只能按照不同的业务代码，由开发人员来衡量设置多长的时间

7.1.4、持有锁时间：

如果锁粒度比较小，时间可以设置长一点，就算执行比较慢，影响面比较小可以接受

7.1.5、难道每次想使用分布式锁的时候都需要下面流程一样，在编码一次？有什么办法能优化吗？

1、先创建一个分布式锁对象；RedisLock redisLock = new RedisLock(lockParam);
2、加锁；Boolean lockFlag = redisLock.lock();
3、finally 解锁；redisLock.unlock();