[大数据]Redis分布式锁

为什么要使用分布式锁

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

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

满足分布式锁的必要条件

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

实现分布式锁的几种方式

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

Redis实现分布式锁的命令

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

可选参数
从 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命令进行加锁。

分布式锁考虑的问题

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

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

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

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

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

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

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

java代码实现Redis分布式锁

先把需要的jar包引入

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

加锁设置参数的实体类

package com.ahysf.common.entity;

import lombok.Data;

/**
 * @ClassName: LockParam
 * @author: 〆、dyh
 * @since: 2022/3/15 15:02
 */

//加锁设置的参数
@Data
public class LockParam {
    //锁的key
    private String lockKey;
    //尝试获得锁的时间（单位：毫秒），默认值：3000毫秒
    private Long tryLockTime;
    //尝试获得锁后，持有锁的时间（单位：毫秒），默认值：5000毫秒
    private Long holdLockTime;

    public LockParam(String lockKey) {
        this(lockKey, 1000 * 3L, 1000 * 5L);
    }


    public LockParam(String lockKey, Long tryLockTime) {
        this(lockKey, tryLockTime, 1000 * 5L);
    }

 

    public LockParam(String lockKey, Long tryLockTime, Long holdLockTime) {
        this.lockKey = lockKey;
        this.tryLockTime = tryLockTime;
        this.holdLockTime = holdLockTime;
    }

}

Redis分布式具体代码实现

package com.ahysf.common.utils;

import com.ahysf.common.entity.LockParam;
import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;

import java.util.Collections;
import java.util.UUID;

/**
 * @ClassName: RedisLock
 * @author: 〆、dyh
 * @since: 2022/3/15 15:11
 */

/**
 * 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;

        this.tryLockEndTime = lockParam.getTryLockTime() + System.currentTimeMillis();
        this.redisLockKey = prefix_key.concat(lockParam.getLockKey());
        this.redisLockValue = UUID.randomUUID().toString().replaceAll("-", "");

        //todo 到时候可以更换获取Jedis实例的实现
        jedis = new Jedis("127.0.0.1", 6379);
    }

    /**
     * 加锁
     *
     * @return 成功返回true，失败返回false
     */
    public boolean lock() {
        while (true) {
            //判断是否超过了，尝试获取锁的时间
            if (System.currentTimeMillis() > tryLockEndTime) {
                return false;
            }
            //尝试获取锁
            holdLockSuccess = tryLock();
            if (Boolean.TRUE.equals(holdLockSuccess)) {
                return true;//获取锁成功
            }

            try {
                //获得锁失败，休眠50毫秒再去尝试获得锁,避免一直请求redis，导致redis cpu飙升
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 执行一次加锁操作：成功返回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分布式锁使用

package com.ahysf.common.utils;

import com.ahysf.common.entity.LockParam;
import lombok.extern.slf4j.Slf4j;

/**
 * @ClassName: TestRedisLock
 * @author: 〆、dyh
 * @since: 2022/3/15 15:36
 */

@Slf4j
public class TestRedisLock {
    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");

        //同时启动线程
        thread1.start();
        thread2.start();

        Thread.sleep(1000 * 20);
        log.info("-----------------我是一条分割线----------------");
        log.info("");
        log.info("");
        log.info("");

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

        Thread.sleep(1000 * 1);//暂停一秒是为了让线程3获的到锁
        Thread thread4 = new Thread(() -> {
            testRedisLock();
        });
        thread4.setName("我是线程4");
        thread4.start();
    }

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

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