Redis应用(二) --分布式锁以及压测介绍

Spring-boot 集成Redis应用(二) --分布式锁以及压测介绍

一.基础环境

jdk 1.8
maven 3.5.3
spring-boot 2.0.4
redis 4.0.11
ApacheBench 2.3

以上工具需要提前安装以及熟悉基本操作,在本文中不会讲解如何安装

二.基本介绍

相信各位小伙伴在学习Redis时,都了解到Redis不仅仅是一个内存中的数据结构存储系统，它可以用作数据库、缓存和消息中间件。上一篇集成应用已经简单的介绍了 Redis作为消息队列配合基于Servlet 3的异步请求处理的简单示例。本篇将介绍Redis在单机部署的场景下的分布式锁。分布式锁的思想来源于Redis官网，下面给出中文翻译相当棒链接，方便大家了解分布式锁。《Redis官方文档》用Redis构建分布式锁。在这就不讲解中心思想了。那么我会以一个模拟秒杀系统的简单Demo来一步一步的展示redis分布式锁的应用。

三.无布式锁时的秒杀代码以及压测结果

组件依赖

由于是无redis锁的情况下的秒杀demo，则只需要引入spring-boot基础依赖即可

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-web</artifactId>
</dependency>

代码

由于是一个简单的秒杀下单的demo代码，那么只需要两个接口，一个是下单接口(order),一个是查询订单接口(query)，Controller层代码如下：

/**
 * @author Neal
 * 测试无分布式锁controller 层
 */
@RestController
@RequestMapping("/order")
public class NoDistributeController {

    //无分布式锁service
    @Autowired
    NoDistributeService redisDistributeService;

    /**
     * 查询剩余订单结果接口
     * @param pid  订单编号
     * @return
     */
    @GetMapping("/query/{pid}")
    public String query(@PathVariable String pid) {
        return redisDistributeService.queryMap(pid);
    }

    /**
     * 下单接口
     * @param pid  订单编号
     * @return
     */
    @GetMapping("/{pid}")
    public String order(@PathVariable String pid) {
        redisDistributeService.order(pid);
        return redisDistributeService.queryMap(pid);
    }
}

service层代码如下:

/**
 * @author Neal
 * 测试无分布式锁service 层
 */
@Service
public class NoDistributeService {

    //模拟商品信息表
    private static Map<String,Integer> products;

    //模拟库存表
    private static Map<String,Integer> stock;

    //模拟订单表
    private static Map<String,String> orders;

    static {
        products = new HashMap<>();
        stock = new HashMap<>();
        orders = new HashMap<>();
        //模拟订单表数据 订单编号 112233 库存 100000
        products.put("112233",100000);
        //模拟库存表数据 订单编号112233 库存100000
        stock.put("112233",100000);
    }

    /**
     * 模拟查询秒杀成功返回的信息
     * @param pid 商品编号
     * @return  返回拼接的秒杀商品结果字符串
     */
    public String queryMap(String pid) {
        return "秒杀商品限量:" +  products.get(pid) + "份,还剩:"+stock.get(pid) +"份,成功下单:"+orders.size() + "人";
    }

    /**
     * 下单方法
     * @param pid  商品编号
     */
    public void order(String pid) {
        //从库存表中获取库存余量
        int stockNum = stock.get(pid);
        //如果库存为0 则输出库存不足
        if(stockNum == 0) {
            System.out.println("商品库存不足");
        }else{ //如果有库存
            //往订单表中插入数据 生成UUID作为用户ID pid
            orders.put(UUID.randomUUID().toString(),pid);
            //线程休眠 模拟其他操作
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //减库存操作
            stock.put(pid,stockNum-1);
        }
    }
}

代码上有注释,我就不过多的啰嗦解释了,那么让我们来模拟秒杀环境，使用 ApacheBench 来模拟并发，来看看结果如何:

进入到 ApacheBench 的bin目录下，我的路径是 Apache24\bin。
输入命令以及参数:ab -n 1000 -c 100 http://127.0.0.1:8080/order/112233 该命令的含义是向指定的URL发送 1000次请求 100 并发量。结果如图

无锁压测图1

无锁压测图2

虽然只耗时了2.596秒,但是处理的结果却不尽如人意。

然后使用查询接口查询一下下单和库存结果是否一致,请求查询接口:http://localhost:8080/order/query/112233。结果如图

无锁查询结果

可以看出下单人数和库存余量明显不符，就这就是无锁时，在高并发环境中会引起的问题。

Redis分布式锁下的秒杀代码以及压测结果

前言

网上看了很多的例子，都是使用redis的SETNX命令来实现的，Redis 官网并不推荐使用SETNX命令，而是推荐使用SET，因为从2.6.12版本以后，Redis对SET命令增加了一系列的选项。

EX seconds – Set the specified expire time, in seconds.
PX milliseconds – Set the specified expire time, in milliseconds.
NX – Only set the key if it does not already exist.
XX – Only set the key if it already exist.
EX seconds – 设置键key的过期时间，单位时秒
PX milliseconds – 设置键key的过期时间，单位时毫秒
NX – 只有键key不存在的时候才会设置key的值
XX – 只有键key存在的时候才会设置key的值

注意: 由于SET命令加上选项已经可以完全取代SETNX, SETEX, PSETEX的功能，所以在将来的版本中，redis可能会不推荐使用并且最终抛弃这几个命令。原文地址

所以本例也是使用上述文章所推荐的加锁解锁方法。

加锁：使用命令 SET resource_name my_random_value NX PX 30000 这个命令的作用是在只有这个key不存在的时候才会设置这个key的值（NX选项的作用），超时时间设为30000毫秒（PX选项的作用）这个key的值设为“my_random_value”。这个值必须在所有获取锁请求的客户端里保持唯一。

解锁： 使用LUA脚本语言

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

这段脚本的意思是:删除这个key当且仅当这个key存在而且值是我期望的那个值。

LUA脚本的原子性原文链接

Redis 使用单个 Lua 解释器去运行所有脚本，并且， Redis 也保证脚本会以原子性(atomic)的方式执行：当某个脚本正在运行的时候，不会有其他脚本或 Redis 命令被执行。这和使用 MULTI / EXEC 包围的事务很类似。在其他别的客户端看来，脚本的效果(effect)要么是不可见的(not visible)，要么就是已完成的(already completed)。

另一方面，这也意味着，执行一个运行缓慢的脚本并不是一个好主意。写一个跑得很快很顺溜的脚本并不难，因为脚本的运行开销(overhead)非常少，但是当你不得不使用一些跑得比较慢的脚本时，请小心，因为当这些蜗牛脚本在慢吞吞地运行的时候，其他客户端会因为服务器正忙而无法执行命令。

组件依赖

代码

1.配置jedis。网上有很多配置jedis的例子,我就给一个简单的配置实现。我使用的是自定义的配置，首先在 application.properties中添加Jedis配置参数

#jedis相关配置
#Redis IP
jedis.host=192.168.56.101
#Redis 端口
jedis.port=6379
#Redis 密码
jedis.password=123456
jedis.timeout=3
jedis.poolMaxTotal=10
jedis.poolMaxIdle=10
jedis.poolMaxWait=3

2.声明自定义配置Bean,使用springboot 注解ConfigurationProperties来加载application.properties中的配置参数。

/**
 * @author Neal
 * 自定义Jedis配置bean
 */
@Component
@ConfigurationProperties(prefix = "jedis")
public class MyJedisBean {

    private String host;

    private int port;

    private String password;

    private int timeout;

    private int poolMaxTotal;

    private int poolMaxIdle;

    private int poolMaxWait;

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public int getTimeout() {
        return timeout;
    }

    public void setTimeout(int timeout) {
        this.timeout = timeout;
    }

    public int getPoolMaxTotal() {
        return poolMaxTotal;
    }

    public void setPoolMaxTotal(int poolMaxTotal) {
        this.poolMaxTotal = poolMaxTotal;
    }

    public int getPoolMaxIdle() {
        return poolMaxIdle;
    }

    public void setPoolMaxIdle(int poolMaxIdle) {
        this.poolMaxIdle = poolMaxIdle;
    }

    public int getPoolMaxWait() {
        return poolMaxWait;
    }

    public void setPoolMaxWait(int poolMaxWait) {
        this.poolMaxWait = poolMaxWait;
    }
}

3.生成JedisPool组件bean

这里就是把JedisPool的相关配置参数配置到JedisPoolConfig并且利用JedisPool的构造方法来声明对象。

/**
 * @author Neal
 * 初始化jedis 连接池
 */
@Component
public class MyJedisConfig {

    /**
     * 自定义jedis配置bean
     */
    @Autowired
    private MyJedisBean myJedisBean;

    @Bean
    public JedisPool jedisPoolFactory() {

        //声明jedispool 配置类
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        jedisPoolConfig.setMaxIdle(myJedisBean.getPoolMaxIdle());
        jedisPoolConfig.setMaxTotal(myJedisBean.getPoolMaxTotal());
        jedisPoolConfig.setMaxWaitMillis(myJedisBean.getPoolMaxWait() *1000);
        /**
         * 利用Jedis的构造方法 生成 jedispool
         */
        JedisPool jedisPool = new JedisPool(jedisPoolConfig,myJedisBean.getHost(),myJedisBean.getPort(),myJedisBean.getTimeout()*1000,myJedisBean.getPassword(),0);
        return jedisPool;
    }

}

4.实现Redis分布式锁方法

该类中只有加锁（redisLock）和解锁（redisUnlock）两个方法。声明的静态变量都是根据Redis原生命令 SET resource_name my_random_value NX PX 30000 声明的命令字符串。在加锁和解锁时 resource_name 对应的是商品ID， my_random_value 对应的是我们用UUID 生成的模拟用户ID。

/**
 * @author Neal
 * 分布式锁
 */
@Component
public class MyRedisLock {
    //Only set the key if it does not already exist.
    private static final String IF_NOT_EXIST = "NX";
    // Set the specified expire time, in milliseconds.
    private static final String SET_EXPIRE_TIME = "PX";
    //超时时间为 500毫秒
    private static final int EXPIRE_TIME = 500;
    //加锁成功后返回的标识
    private static final String ON_LOCK = "OK";
   //LUA 解锁脚本
    private static final String LUA_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    /**
     * 获取配置好的jedis pool 组件
     */
    @Autowired
    private JedisPool jedisPoolFactory;

    /***
     * redis 加锁方法
     * @param id  商品ID
     * @param uuid  模拟用户ID
     * @return 返回 true 加锁成功 false 解锁成功
     */
    public boolean redisLock(String id,String uuid) {
        //从 jedis 连接池中 获取jedis
        Jedis jedis = jedisPoolFactory.getResource();
        boolean locked = false;
        try{
            //使用Jedis 加锁
            locked = ON_LOCK.equals(jedis.set(id,uuid,IF_NOT_EXIST,SET_EXPIRE_TIME,EXPIRE_TIME));
        }finally {
            //将连接放回连接池
            jedis.close();
        }
        return locked;
    }

    /***
     * redis 解锁方法
     * @param id  商品ID
     * @param uuid  模拟用户ID
     * @return  由于是使用LUA脚本，则会保证原子性的特质
     */
    public void redisUnlock(String id,String uuid) {
        //从 jedis 连接池中 获取jedis
        Jedis jedis = jedisPoolFactory.getResource();
        try{
            //使用Jedis 的 eval解锁
            Object result = jedis.eval(LUA_SCRIPT, Collections.singletonList(id),Collections.singletonList(uuid));
            if(1L == (Long)result) {
                System.out.println("客户ID为:《" + uuid + "》   解锁成功！");
            }
        }finally {
            jedis.close();
        }
    }
}

核心的加锁代码已经介绍完了，下面就是关于在秒杀service层加锁与解锁相关的代码了。

5.Controller层

controller层与之前的无锁controller没有变化，还是一样的代码。

/**
 * @author Neal
 * 测试分布式锁controller 层
 */
@RestController
@RequestMapping("/distribute")
public class RedisDistributeController {

    @Autowired
    private RedisDistributeService redisDistributeService;

    @Autowired
    private JedisPool jedisPoolFactory;

    @GetMapping("/query/{pid}")
    public String query(@PathVariable String pid) {
        return redisDistributeService.queryMap(pid);
    }

    @GetMapping("/{pid}")
    public String order(@PathVariable String pid) {
        redisDistributeService.order(pid, UUID.randomUUID().toString());
        return redisDistributeService.queryMap(pid);
    }
}

6.service层

在service层中的下单方法（order）中加入了加锁与解锁的操作。

/**
 * @author Neal
 * 测试分布式锁service 层
 */
@Service
public class RedisDistributeService {


    //模拟商品信息表
    private static Map<String,Integer> products;

    //模拟库存表
    private static Map<String,Integer> stock;

    //模拟订单表
    private static Map<String,String> orders;

    //redis 锁组件
    @Autowired
    MyRedisLock myRedisLock;

    static {
        products = new HashMap<>();
        stock = new HashMap<>();
        orders = new HashMap<>();
        products.put("112233",100000);
        stock.put("112233",100000);
    }

    /**
     * 模拟查询秒杀成功返回的信息
     * @param pid 商品名称
     * @return
     */
    public String queryMap(String pid) {
        return "秒杀商品限量:" +  products.get(pid) + "份,还剩:"+stock.get(pid) +"份,成功下单:"+orders.size() + "人";
    }

    /**
     * 下单方法
     * @param pid  商品名称
     */
    public void order(String pid,String uuid) {

        //redis 加锁
        if(!myRedisLock.redisLock(pid,uuid)) {  //如果没获得锁则直接返回，不执行下面的代码
            System.out.println("客户ID为:《"+ uuid +"》未获得锁");
            return;
        }

        System.out.println("客户ID为:《"+ uuid +"》获得锁");
        //从库存表中获取库存余量
        int stockNum = stock.get(pid);
        if(stockNum == 0) {
            System.out.println("商品库存不足");
        }else{
            //往订单表中插入数据
            orders.put(uuid,pid);
            //线程休眠 模拟其他操作
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //减库存操作
            stock.put(pid,stockNum-1);
        }

        //redis 解锁
        myRedisLock.redisUnlock(pid,uuid);
    }
}

<u>代码已经介绍完了那么下面开始压测，并看看结果是否跟没加锁的代码有区别</u>

7.压测

压测方式跟上面讲述的一样，我就不在啰嗦了，直接上结果图。

有锁压测图1

有锁压测图2

结果

由图中可以看出，同样的压测命令，秒杀的结果却只有13个秒杀成功，但是库存余量与秒杀的数量是对应的上的，不会出现库存与秒杀数量不一致问题。

结论：

在写Redis锁的时候看了很多前辈的博文以及教学视频，发现之前用的都是基于SETNX，解锁也是各有千秋，最后还是参考了 Redis的官方实践。
DEMO代码地址
 Redis命令大全

最后编辑于：2019.02.07 18:07:26

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 206,378评论 6赞 481
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 88,356评论 2赞 382
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 152,702评论 0赞 342
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 55,259评论 1赞 279
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 64,263评论 5赞 371
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 49,036评论 1赞 285
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,349评论 3赞 400
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,979评论 0赞 259
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 43,469评论 1赞 300
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,938评论 2赞 323
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,059评论 1赞 333
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,703评论 4赞 323
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,257评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,262评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,485评论 1赞 262
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,501评论 2赞 354
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,792评论 2赞 345

Redis应用(二) --分布式锁以及压测介绍

Spring-boot 集成Redis应用(二) --分布式锁以及压测介绍

一.基础环境

二.基本介绍

三.无布式锁时的秒杀代码以及压测结果

组件依赖

代码

Redis分布式锁下的秒杀代码以及压测结果

前言

LUA脚本的原子性 原文链接

组件依赖

代码

结论：

推荐阅读更多精彩内容

LUA脚本的原子性原文链接