【Redis】事务秒杀案例

一、背景

在日常购物时，经常会有商家开展限时秒杀活动，我们如何使用redis来实现这种场景呢

二、业务代码

首先我们可以想到的是，我们可以把商品剩余数量和成功秒杀商品的用户id放在redis中

下面是我们的业务代码

package com.decade.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Controller;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import redis.clients.jedis.Jedis;

import java.util.UUID;

@Controller
@Slf4j
public class IndexController {
    @PostMapping(value = "/testSecondsKill")
    @ResponseBody
    public void testSecondsKill(@RequestParam(value = "prodId") String prodId) {
        try {
            secondKill(prodId, UUID.randomUUID().toString());
        } catch (Exception exception) {
            log.error("请求出错", exception);
        }
    }

    private boolean secondKill(String prodId, String uid) {
        // 判断商品id和用户id是否为空
        if (!(StringUtils.hasLength(prodId) && StringUtils.hasLength(uid))) {
            System.out.println("商品id或用户id为空");
            return false;
        }

        // 构建redis中存放秒杀成功用户的key和商品剩余库存的key
        String productKey = "prod:" + prodId + "-key";
        String userKey = "user:" + prodId + "-key";

        // 连接redis
        Jedis jedisClient = new Jedis("192.168.153.128", 6379);

        // 判断是否已经开始秒杀，即判断商品库存是否还是null
        if (null == jedisClient.get(productKey)) {
            System.out.println("秒杀未开始！");
            // 关闭redis连接
            jedisClient.close();
            return false;
        }

        // 判断用户是否重复秒杀
        if (jedisClient.sismember(userKey, uid)) {
            System.out.println("请勿重复秒杀！");
            jedisClient.close();
            return false;
        }

        // 判断是否还存在商品库存，即商品数量少于1（或等于0）
        if (Integer.parseInt(jedisClient.get(productKey)) < 1) {
            System.out.println("秒杀已结束");
            jedisClient.close();
            return false;
        }

        // 如果还存在商品库存，那么将用户id放入redis中，并且对商品库存做递减操作
        jedisClient.sadd(userKey, uid);
        jedisClient.decr(productKey);
        System.out.println("秒杀成功！");
        jedisClient.close();
        return true;
    }
}

在发起请求前，我们先去redis中初始化商品剩余数量，设置为10

然后通过postman发起请求，如果涉及到用户登录验证，那么需要在请求头的Cookie中加入JSESSIONID，然后请求http://localhost:8080/testSecondsKill

控制台输出结果如下

redis中各key对应的value变化如下

127.0.0.1:6379> keys *
(empty array)
# 初始化商品剩余数量
127.0.0.1:6379> set prod:huawei-key 10
OK
127.0.0.1:6379> get prod:huawei-key
"10"
127.0.0.1:6379>
# 抢购一次之后的剩余数量
127.0.0.1:6379> get prod:huawei-key
"9"
127.0.0.1:6379> keys *
1) "user:huawei-key"
2) "/testSecondsKill"
3) "prod:huawei-key"
127.0.0.1:6379>
# 查看成功秒杀用户id
127.0.0.1:6379> smembers user:huawei-key
1) "b425c3ee-ba2d-417e-876d-8cfc5f298900"
127.0.0.1:6379>
127.0.0.1:6379>
127.0.0.1:6379> smembers user:huawei-key
 1) "728ac609-2de4-45f8-9ed9-55d22014ccc3"
 2) "443e8ea7-ca00-4680-b56b-a4061221edfa"
 3) "b425c3ee-ba2d-417e-876d-8cfc5f298900"
 4) "808a1f50-7696-471a-bebe-8091ed92fa97"
 5) "09edc482-50ee-4d5c-9c8b-691fb68775f0"
 6) "1a7f0c06-9e41-4844-873c-f04ee293ef87"
 7) "bb9e27cd-facf-4eb1-b5dd-533507336ec6"
 8) "745d4d3c-8bf8-41de-b339-24eb9ec21eea"
 9) "c4357b6e-e8a9-4d4c-b9da-27cae191fa50"
10) "6a62ea01-da9b-46a9-9174-5f3304463a6b"
127.0.0.1:6379>
127.0.0.1:6379>
127.0.0.1:6379> get prod:huawei-key
"0"
127.0.0.1:6379>
127.0.0.1:6379>
127.0.0.1:6379> keys *
1) "user:huawei-key"
2) "/testSecondsKill"
3) "prod:huawei-key"
127.0.0.1:6379>

三、使用ab工具模拟并发

在实际应用场景中，请求不可能是一个个过来的，所以我们需要使用并发模拟工具来看下高并发场景下会发生什么问题

1、安装相关工具
我的虚拟机系统是CentOS7，需要手动安装

yum install httpd-tools

然后我们可以使用ab --help命令简单查看一下如何使用

• -n：请求次数
• -c：并发量，也就是同一时间发送给服务器的请求次数
• -p：如果请求类型是POST，且请求体携带参数，那么就需要把参数放到文件中，而且还要设置下方的-T参数
• -T：如果使用的是POST/PUT类型的请求，那么需要将请求头中的内容类型设置为application/x-www-form-urlencoded，默认值是text/plain
  

大概格式为

ab -n 请求次数 -c 并发数 -p 存放参数的文件路径 -T x-www-form-urlencoded 请求地址

然后我们在某个路径下创建一个文件存放参数

然后我们使用命令ab -n 1000 -c 100 -p /opt/decade/param.txt -T application/x-www/form-urlencoded http://192.168.0.115:8080/testSecondsKill去模拟并发调用

我们发现，控制台输出的内容和我们预计的不一样，商品在秒杀结束之后还能被抢购成功
而且redis中的商品剩余数量的值变为了负数，这明显是不符合常理的

另外，如果并发量太大的情况下，可能还会出现redis连接超时的情况，因为redis无法同时处理这么多请求，当请求等待时间过长，就会报错redis连接超时

四、超卖和超时问题的解决

上面的案例中，我们遇到了超卖和超时的问题，那么我们应该如何解决呢？

• 解决连接超时问题：使用连接池
  下面我们通过代码样例了解一下

package com.decade.util;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import java.time.Duration;

public class JedisPoolUtil {

    private static volatile JedisPool jedisPool = null;

    private JedisPoolUtil() {

    }

    /**
     * 获取jedis连接池
     * @return 返回jedis连接池
     */
    public static JedisPool getJedisPool() {
        if (null == jedisPool) {
            synchronized (JedisPoolUtil.class) {
                if (null == jedisPool) {
                    JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
                    jedisPoolConfig.setMaxTotal(200);
                    jedisPoolConfig.setMaxIdle(32);
                    jedisPoolConfig.setMaxWait(Duration.ofMillis(100 * 1000));
                    jedisPoolConfig.setBlockWhenExhausted(true);
                    jedisPoolConfig.setTestOnBorrow(true);

                    jedisPool = new JedisPool(jedisPoolConfig, "192.168.153.128", 6379, 60000);
                }
            }
        }
        return jedisPool;
    }

    /**
     * 释放连接池资源
     * @param jedisPool 连接池
     * @param jedis jedis连接
     */
    public static void release(JedisPool jedisPool, Jedis jedis) {
        if (null != jedis) {
            jedisPool.returnResource(jedis);
        }
    }
}

然后我们就可以通过这个工具类来获取jedis连接，使用完成之后，调用工具类中的方法释放连接

// 连接redis
final JedisPool jedisPool = JedisPoolUtil.getJedisPool();
final Jedis jedisPoolResource = jedisPool.getResource();
...

// 释放连接
JedisPoolUtil.release(jedisPool, jedisPoolResource);

• 解决超卖问题：使用事务（乐观锁）
  我们将上面的业务代码进行改造，开启事务

private boolean secondKill(String prodId, String uid) {
    ...

    // 开启对库存剩余数量这个key的监视
    jedisClient.watch(productKey);

    // 判断是否已经开始秒杀，即判断商品库存是否还是null
    if (null == jedisClient.get(productKey)) {
        System.out.println("秒杀未开始！");
        // 关闭redis连接
        jedisClient.close();
        return false;
    }

    ...

    // 判断是否还存在商品库存，即商品数量少于1（或等于0）
    if (Integer.parseInt(jedisClient.get(productKey)) < 1) {
        System.out.println("秒杀已结束");
        jedisClient.close();
        return false;
    }

    // 使用事务
    final Transaction multi = jedisClient.multi();

    // 组队操作
    multi.sadd(userKey, uid);
    multi.decr(productKey);

    // 执行
    final List<Object> results = multi.exec();

    // 对返回结果做一个判断
    if (null == results || 0 == results.size()) {
        System.out.println("秒杀失败！");
        jedisClient.close();
        return false;
    }
    System.out.println("秒杀成功！");
    jedisClient.close();
    return true;
}

五、库存遗留问题

当我们使用乐观锁去解决超卖问题时，我们可以发现，当初始库存变大时（假设为500）
使用并发模拟工具进行测试发现，最后还剩余一部分没有卖完
可是我们的请求次数是大于初始库存的

分析：当第一个用户秒杀完成之后，会对原来的数据进行修改，版本号也会随之改变
这样，后面到达的1999个请求发现版本号发生了变化，修改请求就被取消了

那么，如何解决这种问题呢？使用LUA脚本(Redis2.6版本之后)

1、好处

• 减少网络开销，可以将多个请求通过脚本的形式一次发送，减少网络时延。
• 原子操作，Redis会将整个脚本作为一个整体执行，中间不会被其他请求插入，因此在脚本运行过程中无需担心会出现竞态条件，无需使用事务
• 复用，客户端发送的脚本会永久存在redis中，这样其他客户端可以复用这一脚本，而不需要使用代码完成相同的逻辑

这样，就类似于使用了悲观锁，既解决了超卖问题，又能避免库存剩余问题

首先我们要看下LUA脚本怎么写

local userid=KEYS[1]; 
local prodid=KEYS[2];
local qtkey="prod:"..prodid.."-key";
local usersKey="user:"..prodid."-key'; 
local userExists=redis.call("sismember",usersKey,userid);
if tonumber(userExists)==1 then 
  return 2;
end
local num= redis.call("get" ,qtkey);
if tonumber(num)<=0 then 
  return 0; 
else 
  redis.call("decr",qtkey);
  redis.call("sadd",usersKey,userid);
end
return 1;

然后我们需要在Java代码中使用

package com.decade.util;

import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.io.IOException;

@Slf4j
public class JedisScriptUtil {

    static String secKillScript ="local userid=KEYS[1];\r\n" +
        "local prodid=KEYS[2];\r\n" +
        "local qtkey='prod:'..prodid..\"-key\";\r\n" +
        "local usersKey='user:'..prodid..\"-key\";\r\n" +
        "local userExists=redis.call(\"sismember\",usersKey,userid);\r\n" +
        "if tonumber(userExists)==1 then \r\n" +
        "   return 2;\r\n" +
        "end\r\n" +
        "local num= redis.call(\"get\" ,qtkey);\r\n" +
        "if tonumber(num)<=0 then \r\n" +
        "   return 0;\r\n" +
        "else \r\n" +
        "   redis.call(\"decr\",qtkey);\r\n" +
        "   redis.call(\"sadd\",usersKey,userid);\r\n" +
        "end\r\n" +
        "return 1" ;

    public static boolean doSecKill(String prodId, String uid) throws IOException {
        JedisPool jedispool = JedisPoolUtil.getJedisPool();
        Jedis jedis = jedispool.getResource();

        // Redis Script Load 命令用于将脚本 script 添加到脚本缓存中，但并不立即执行这个脚本
        String sha1 = jedis.scriptLoad(secKillScript);

        // Redis Evalsha 命令根据给定的 sha1 校验码，执行缓存在服务器中的脚本
        // sha1:通过 SCRIPT LOAD生成的sha1校验码 keyCount:键的个数 uid,prodId:要操作的key
        Object result = jedis.evalsha(sha1, 2, uid, prodId);

        String reString = String.valueOf(result);
        if ("0".equals( reString )  ) {
            System.err.println("已抢空！！");
        }else if("1".equals( reString )  )  {
            System.out.println("抢购成功！！！！");
        }else if("2".equals( reString )  )  {
            System.err.println("该用户已抢过！！");
        }else{
            System.err.println("抢购异常！！");
        }
        jedis.close();
        return true;
    }
}

然后在接口处调用该方法

final boolean secKillResult = JedisScriptUtil.doSecKill(prodId, UUID.randomUUID().toString());

经过验证得知，在使用连接池和LUA脚本后
库存遗留、超卖和连接超时的问题都解决了

如有错误，欢迎指正！！！