Redis分布式锁解决方案

我们知道分布式锁的特性是排他、避免死锁、高可用。分布式锁的实现可以通过数据库的乐观锁(通过版本号)或者悲观锁(通过for update)、Redis的setnx()命令、Zookeeper(在某个持久节点添加临时有序节点,判断当前节点是否是序列中最小的节点,如果不是则监听比当前节点还要小的节点。如果是,获取锁成功。当被监听的节点释放了锁(也就是被删除),会通知当前节点。然后当前节点再尝试获取锁,如此反复)

redis.png

本篇文章,主要讲如何用Redis的形式实现分布式锁。后续文章会讲解热点KEY读取,缓存穿透和缓存雪崩的场景和解决方案、缓存更新策略等等知识点,理论知识点较多。

Redis配置

我的redis配置如下

spring.redis.host=
spring.redis.port=6379
#reids超时连接时间
spring.redis.timeout=100000
spring.redis.password=
#连接池最大连接数
spring.redis.pool.max-active=10000
#连接池最大空闲数
spring.redis.pool.max-idle=1000
#连接池最大等待时间
spring.redis.pool.max-wait=10000
@Component
@Getter
@Setter
@ConfigurationProperties(prefix = "spring.redis")
public class RedisConfig {

    @Value("${spring.redis.host}")
    private String host;

    @Value("${spring.redis.port}")
    private int port;

    @Value("${spring.redis.password}")
    private String password;

    @Value("${spring.redis.timeout}")
    private int timeout;

    @Value("${spring.redis.pool.max-active}")
    private int poolMaxActive;

    @Value("${spring.redis.pool.max-idle}")
    private int poolMaxIdle;

    @Value("${spring.redis.pool.max-wait}")
    private int poolMaxWait;
}
@Component
public class RedisPoolFactory {

    @Autowired
    private RedisConfig redisConfig;

    @Bean
    public JedisPool jedisPoolFactory() {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxIdle(redisConfig.getPoolMaxIdle());
        poolConfig.setMaxTotal(redisConfig.getPoolMaxActive());
        poolConfig.setTestOnBorrow(true);
        poolConfig.setMaxWaitMillis(redisConfig.getPoolMaxWait());
        JedisPool jp = new JedisPool(poolConfig, redisConfig.getHost(), redisConfig.getPort(),
                redisConfig.getTimeout(), redisConfig.getPassword(), 0);
        return jp;
    }

}

为了区分不同模块的key,我抽象出了一个KeyPrefix接口和BasePrefix类。

public interface KeyPrefix {

    int expireSeconds();

    String getPrefix();
}
/**
 * @author cmazxiaoma
 * @version V1.0
 * @Description: TODO
 * @date 2018/5/10 12:35
 */
public abstract class BasePrefix implements KeyPrefix {

    private int expireSeconds;

    private String prefix;

    public BasePrefix(int expireSeconds, String prefix) {
        this.expireSeconds = expireSeconds;
        this.prefix = prefix;
    }

    public BasePrefix(String prefix) {
        this(0, prefix);
    }

    @Override
    public int expireSeconds() {
        return expireSeconds;
    }

    @Override
    public String getPrefix() {
        String className = getClass().getSimpleName();
        return className + ":" + prefix;
    }

}

分布式锁分析与编码

下面进入正文。因为分布式系统之间是不同进程的,单机版的锁无法满足要求。所以我们可以借助中间件Redis的setnx()命令实现分布式锁。setnx()命令只会对不存在的key设值,返回1代表获取锁成功。对存在的key设值,会返回0代表获取锁失败。这里的value是System.currentTimeMillis() (获取锁的时间)+锁持有的时间。我这里设置锁持有的时间是200ms,实际业务执行的时间远比这200ms要多的多,持有锁的客户端应该检查锁是否过期,保证锁在释放之前不会过期。因为客户端故障的情况可能是很复杂的。比如现在有A,B俩个客户端。A客户端获取了锁,执行业务中做了骚操作导致阻塞了很多,时间应该远远超过200ms,当A客户端从阻塞状态下恢复继续执行业务代码时,A客户端持有的锁由于过期已经被其他客户端占有。这时候A客户端执行释放锁的操作,那么有可能释放掉其他客户端的锁。

我这里设置的客户端等待锁的时间是200ms。这里通过轮询的方式去让客户端获取锁。如果客户端在200ms之内没有锁的话,直接返回false。实际场景要设置合适的客户端等待锁的时间,避免消耗CPU资源。

如果获取锁的逻辑只有这三行代码的话,会造成死循环,明显不符合分布式锁的特性。

                if (jedis.setnx(realKey, value) == 1) {
                    return true;
                }

所以,我们要加上锁过期,然后获取锁的策略。通过realKey获取当前的currentValue。currentValue也就是获取锁的时间 + 锁持有的时间。 如果currentValue不等于null 且 currentValue 小于当前时间,说明锁已经过期。这时候如果突然来了C,D两个客户端获取锁的请求,不就让C,D两个客户端都获取锁了吗。如果防止这种现象发生,我们采用getSet()命令来解决。getSet(key,value)的命令会返回key对应的value,然后再把key原来的值更新为value。也就是说getSet()返回的是已过期的时间戳。如果这个已过期的时间戳等于currentValue,说明获取锁成功。

假设客户端A一开始持有锁,保存在redis中的value(时间戳)等于T1。 这时候客户端A的锁已经过期,那么C,D客户端就可以开始争抢锁了。currentValue是T1,C客户端的value是T2,D客户端的value是T3。首先C客户端进入到String oldValue = jedis.getSet(realKey, value);这行代码,获得的oldValue是T1,同时也会把realKey对应的value更新为T2。再执行后续的代码,oldValue等于currentValue,那么客户端C获取锁成功。接着D客户端也执行到了String oldValue = jedis.getSet(realKey, value);这行代码,获取的oldValue是T2,同时也会把realKey对应的value更新为T3。由于oldValue不等于currentValue,那么客户端D获取锁失败。

    public boolean lock(KeyPrefix prefix, String key, String value) {
        Jedis jedis = null;
        Long lockWaitTimeOut = 200L;
        Long deadTimeLine = System.currentTimeMillis() + lockWaitTimeOut;

        try {
            jedis = jedisPool.getResource();
            String realKey = prefix.getPrefix() + key;

            for (;;) {
                if (jedis.setnx(realKey, value) == 1) {
                    return true;
                }

                String currentValue = jedis.get(realKey);

                // if lock is expired
                if (!StringUtils.isEmpty(currentValue) &&
                        Long.valueOf(currentValue) < System.currentTimeMillis()) {
                    // gets last lock time
                    String oldValue = jedis.getSet(realKey, value);

                    if (!StringUtils.isEmpty(oldValue) && oldValue.equals(currentValue)) {
                        return true;
                    }
                }

                lockWaitTimeOut = deadTimeLine - System.currentTimeMillis();

                if (lockWaitTimeOut <= 0L) {
                    return false;
                }
            }
        } finally {
            returnToPool(jedis);
        }
    }

我们讲解了获取的逻辑,接着讲讲释放锁的逻辑。我们在这里加上!StringUtils.isEmpty(currentValue) && value.equals(currentValue)判断是为了防止释放了不属于当前客户端的锁。还是举个例子,如果没有这个逻辑,A客户端调用unlock()方法之前,锁突然就过期了。这时候B客户端发现锁过期了,立马获取了锁。然后A客户端接着调用unlock()方法,却释放了原本属于B客户端的锁。

    public void unlock(KeyPrefix prefix, String key, String value) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            String realKey = prefix.getPrefix() + key;
            String currentValue = jedis.get(realKey);

            if (!StringUtils.isEmpty(currentValue)
                    && value.equals(currentValue)) {
                jedis.del(realKey);
            }
        } catch (Exception ex) {
            log.info("unlock error");
        } finally {
            returnToPool(jedis);
        }
    }

编码RedisController,模拟商品秒杀操作。测试分布式锁是否可行。

/**
 * @author cmazxiaoma
 * @version V1.0
 * @Description: TODO
 * @date 2018/8/28 9:27
 */
@RestController
@RequestMapping("/redis")
public class RedisController {

    private static LongAdder longAdder = new LongAdder();
    private static Long LOCK_EXPIRE_TIME = 200L;
    private static Long stock = 10000L;

    @Autowired
    private RedisService redisService;

    static {
        longAdder.add(10000L);
    }

    @GetMapping("/v1/seckill")
    public String seckillV1() {
        Long time = System.currentTimeMillis() + LOCK_EXPIRE_TIME;
        if (!redisService.lock(SeckillKeyPrefix.seckillKeyPrefix, "redis-seckill", String.valueOf(time))) {
            return "人太多了,换个姿势操作一下";
        }

        if (longAdder.longValue() == 0L) {
            return "已抢光";
        }

        doSomeThing();

        if (longAdder.longValue() == 0L) {
            return "已抢光";
        }

        longAdder.decrement();

        redisService.unlock(SeckillKeyPrefix.seckillKeyPrefix, "redis-seckill", String.valueOf(time));

        Long stock = longAdder.longValue();
        Long bought = 10000L - stock;
        return "已抢" + bought + ", 还剩下" + stock;
    }

    @GetMapping("/detail")
    public String detail() {
        Long stock = longAdder.longValue();
        Long bought = 10000L - stock;
        return "已抢" + bought + ", 还剩下" + stock;
    }

    @GetMapping("/v2/seckill")
    public String seckillV2() {
        if (longAdder.longValue() == 0L) {
            return "已抢光";
        }

        doSomeThing();

        if (longAdder.longValue() == 0L) {
            return "已抢光";
        }

        longAdder.decrement();

        Long stock = longAdder.longValue();
        Long bought = 10000L - stock;
        return "已抢" + bought + ", 还剩下" + stock;
    }

    @GetMapping("/v3/seckill")
    public String seckillV3() {
        if (stock == 0) {
            return "已抢光";
        }

        doSomeThing();
        stock--;

        Long bought = 10000L - stock;
        return "已抢" + bought + ", 还剩下" + stock;
    }


    public void doSomeThing() {
        try {
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }
    }
}

http://localhost:8081/redis/v1/seckill进行压测,我使用的压测工具是ab测试工具。这里用10000个并发用户,20000个请求来进行压测。

ab -c 10000 -n 20000 http://localhost:8081/redis/v1/seckill

压测结果如下:

E:\cmazxiaoma_download\httpd-2.4.34-o102o-x64-vc14\Apache24\bin>ab -c 10000 -n 2
0000 http://localhost:8081/redis/v1/seckill
This is ApacheBench, Version 2.3 <$Revision: 1826891 $>
Copyright 1996 Adam Twiss, Zeus Technology Ltd, http://www.zeustech.net/
Licensed to The Apache Software Foundation, http://www.apache.org/

Benchmarking localhost (be patient)
Completed 2000 requests
Completed 4000 requests
Completed 6000 requests
Completed 8000 requests
Completed 10000 requests
Completed 12000 requests
Completed 14000 requests
Completed 16000 requests
Completed 18000 requests
Completed 20000 requests
Finished 20000 requests


Server Software:
Server Hostname:        localhost
Server Port:            8081

Document Path:          /redis/v1/seckill
Document Length:        22 bytes

Concurrency Level:      10000
Time taken for tests:   108.426 seconds
Complete requests:      20000
Failed requests:        19991
   (Connect: 0, Receive: 0, Length: 19991, Exceptions: 0)
Total transferred:      3420218 bytes
HTML transferred:       760218 bytes
Requests per second:    184.46 [#/sec] (mean)
Time per request:       54213.000 [ms] (mean)
Time per request:       5.421 [ms] (mean, across all concurrent requests)
Transfer rate:          30.80 [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:        0    0   6.3      0     549
Processing:  2393 36477 16329.1  45101   90269
Waiting:      182 36435 16351.4  45046   90267
Total:       2393 36477 16329.0  45101   90269

Percentage of the requests served within a certain time (ms)
  50%  45101
  66%  47680
  75%  49136
  80%  50392
  90%  53200
  95%  53743
  98%  54510
  99%  56014
 100%  90269 (longest request)

我们再来看看是否有超卖现象,貌似还是正常。


回溯分析

我打开RedisDesktopManager查看db0的key信息时,发现还有一个key没有删除掉。说明我们写的unlock()方法在1w并发用户,2w请求下还是存在问题。

image.png

仔细推敲自己之前写的代码发现(还是拿上面的例子说事),客户端D虽然获取锁失败,但是之前进行了String oldValue = jedis.getSet(realKey, value);操作,还是成功的更新了realKey对应的value。我们进行unlock()操作时,释放客户端的锁是根据value来标识当前客户端的。一开始客户端C的value是T2,由于客户端D的getSet()操作,覆盖掉了客户端C的value,让其更新成T3。由于value.equals(currentValue)条件不成立,所以不会执行到jedis.del(realKey)

其实lock()方法也经不起推敲: 1.分布式各个系统时间不一致,如果要这样做,只能进行时间同步。 2.当某个客户端锁过期时,多个客户端开始争抢锁。虽然最后只有一个客户端能成功锁,但是获取锁失败的客户端能覆盖获取锁成功客户端的过期时间。 3.当客户端的锁过期时间被覆盖,会造成锁不具有标识性,会造成客户端没有释放锁。

所以我们要重写lock与unlock()的逻辑,看到网上已经有很多的解决方案。(不过也有很多错误案例)

我们可以通过redis的set(key,value,NX,EX,timeout)合并普通的set()和expire()操作,使其具有原子性。

 /**
   * Set the string value as value of the key. The string can't be longer than 1073741824 bytes (1
   * GB).
   * @param key
   * @param value
   * @param nxxx NX|XX, NX -- Only set the key if it does not already exist. XX -- Only set the key
   *          if it already exist.
   * @param expx EX|PX, expire time units: EX = seconds; PX = milliseconds
   * @param time expire time in the units of <code>expx</code>
   * @return Status code reply
   */
  public String set(final String key, final String value, final String nxxx, final String expx,
      final long time) {
    checkIsInMultiOrPipeline();
    client.set(key, value, nxxx, expx, time);
    return client.getStatusCodeReply();
  }

通过set(key,value,NX,EX,timeout)方法,我们就可以轻松实现分布式锁。值得注意的是这里的value作为客户端锁的唯一标识,不能重复。

    public boolean lock1(KeyPrefix prefix, String key, String value, Long lockExpireTimeOut,
                         Long lockWaitTimeOut) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            String realKey = prefix.getPrefix() + key;
            Long deadTimeLine = System.currentTimeMillis() + lockWaitTimeOut;

            for (;;) {
                String result = jedis.set(realKey, value, "NX", "PX", lockExpireTimeOut);

                if ("OK".equals(result)) {
                    return true;
                }

                lockWaitTimeOut = deadTimeLine - System.currentTimeMillis();

                if (lockWaitTimeOut <= 0L) {
                    return false;
                }
            }
        } catch (Exception ex) {
            log.info("lock error");
        } finally {
            returnToPool(jedis);
        }

        return false;
    }

我们可以使用lua脚本合并get()和del()操作,使其具有原子性。一切大功告成。

    public boolean unlock1(KeyPrefix prefix, String key, String value) {

        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            String realKey = prefix.getPrefix() + key;

            String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

            Object result = jedis.eval(luaScript, Collections.singletonList(realKey),
                    Collections.singletonList(value));

            if ("1".equals(result)) {
                return true;
            }

        } catch (Exception ex) {
            log.info("unlock error");
        } finally {
            returnToPool(jedis);
        }
        return false;

    }

尾言

大家好,我是cmazxiaoma(寓意是沉梦昂志的小马),感谢各位阅读本文章。 小弟不才。 如果您对这篇文章有什么意见或者错误需要改进的地方,欢迎与我讨论。 如果您觉得还不错的话,希望你们可以点个赞。 希望我的文章对你能有所帮助。 有什么意见、见解或疑惑,欢迎留言讨论。

最后送上:心之所向,素履以往。生如逆旅,一苇以航。

saoqi.png

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏java 成神之路

基于ReentrantLock发生死锁的解决方案

33560
来自专栏谈补锅

CocoaAsyncSocket + Protobuf 处理粘包和拆包问题

      在上一篇文章《iOS之ProtocolBuffer搭建和示例demo》分享环境的搭建, 我们和服务器进行IM通讯用了github有名的框架Cocoa...

24740
来自专栏遊俠扎彪

C++ 的 Name Mangling

编程语言组织程序,都有一定的可见范围,比如Java的包、C/C++的文件。就像我们平时使用的文件夹一样,有了这些组织机制,变量、函数的命名在一定程度上就可以重复...

75500
来自专栏desperate633

Java并发之“饥饿”和“公平锁”(Starvation and Fairness)java中发生线程饥饿的原因java中实现公平锁公平锁性能考虑

如果一个线程的cpu执行时间都被其他线程抢占了,导致得不到cpu执行,这种情况就叫做“饥饿”,这个线程就会出现饥饿致死的现象,因为永远无法得到cpu的执行。解决...

15610
来自专栏软件工程师成长笔记

IE、FireFox、Chrome浏览器中关于URL传参中文乱码,解决兼容性问题!

前台用url传值中文,后台用request.getParameter接收参数。在Firefox,Chrome等浏览器中没有问题。但用IE浏览器就又会出现参数中文...

79020
来自专栏芋道源码1024

分布式作业 Elastic-Job-Lite 源码分析 —— 注册中心

ZookeeperRegistryCenter,基于 Zookeeper 注册中心。从上面的类图可以看到,ZookeeperRegistryCenter 实现 ...

19520
来自专栏大内老A

认识ASP.NET MVC的5种AuthorizationFilter

在总体介绍了筛选器及其提供机制(《深入探讨ASP.NET MVC的筛选器》)之后,我们按照执行的先后顺序对四种不同的筛选器进行单独介绍,首先来介绍最先执行的Au...

26060
来自专栏逸鹏说道

C# 温故而知新: 线程篇(四)

线程同步篇 (中):同步工具类的介绍 1 上篇回顾 2 继续介绍基元内核模式中的 monitor类 3 同步句柄:WaitHandle 4 EventW...

32460
来自专栏我的小碗汤

来看三个问题

是否允许在HTTP请求时,返回原始请求体数据字节,默认为false(GET or HEAD or 上传文件请求除外)。

13710
来自专栏Jimoer

JVM学习记录-类加载器

JVM设计团队把类加载阶段中的“通过一个类的全限定名来获取描述此类的二进制字节流”这个动作放到Java虚拟机外面去实现,以便让应用程序自己决定如何去获取所需要的...

8210

扫码关注云+社区

领取腾讯云代金券