前言

秒杀场景为了防止库存超卖有很多种方式，数据库锁（主要用行锁）、分布式锁（redis或zk）、redis+lua脚本实现原子性操作等，使用数据库锁方案适合一些访问量不大的程序，使用分布式锁性能上可能会有所提升主要看业务处理方案，如果锁粒度比较大处理业务逻辑过多性能也不太高甚至比只用数据库锁性能更差，redis+lua脚本能比较好的解决超卖问题，同样的也需要自己处理好业务逻辑，这里只对redis+lua脚本怎么实现扣减库存做编写，业务上怎么处理自己发挥即可。

为什么使用Redisson

redisson是一个redis的客户端，它拥有丰富的功能，而且支持redis的各种模式，什么单机，集群，哨兵的都支持，各种各样的分布式锁实现，什么分布式重入锁，红锁，读写锁，然后它操作redis的常用数据结构就跟操作jdk的各种集合一样简单，底层实现中也是使用了大量的lua脚本操作redis。

项目搭建

maven配置

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.3.RELEASE</version>
    </parent>

	<properties>
        <java.version>1.8</java.version>
        <lombok.version>1.18.12</lombok.version>
        <redisson.version>3.14.1</redisson.version>
    </properties>
	
	<dependencies>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>${lombok.version}</version>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>${redisson.version}</version>
        </dependency>
    </dependencies>

编写Redisson配置类

import lombok.Data;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.StringCodec;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 单机连接
 */
@Data
@Configuration
public class RedissonConfig {
    @Value("${spring.redis.host}")
    private String host;
    @Value("${spring.redis.port}")
    private String port;
    @Value("${spring.redis.password}")
    private String password;
    @Value("${spring.redis.database}")
    private Integer database;
    @Value("${spring.redis.connectTimeout}")
    private Integer connectTimeout;
    @Value("${spring.redis.timeout}")
    private Integer timeout;
    @Value("${spring.redis.connectionPoolSize}")
    private Integer connectionPoolSize;
    @Value("${spring.redis.connectionMinimumIdleSize}")
    private Integer connectionMinimumIdleSize;
    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        String address = new StringBuilder("redis://").append(host).append(":").append(port).toString();
        SingleServerConfig singleServerConfig = config.useSingleServer();//使用单机redis
        singleServerConfig.setAddress(address);
        if (null != password && !"".equals(password.trim())) {
            singleServerConfig.setPassword(password);
        }
        singleServerConfig.setDatabase(database);
        singleServerConfig.setConnectTimeout(connectTimeout); //连接超时时间
        singleServerConfig.setTimeout(timeout); //命令执行等待时间
        singleServerConfig.setConnectionPoolSize(connectionPoolSize); //连接池
        singleServerConfig.setConnectionMinimumIdleSize(connectionMinimumIdleSize); //最小空闲连接数
        //设置String序列化
        //这里如果使用JsonJacksonCodec编码在使用hash结构时对应hash中的key会被套上一个双引号使用redis命令 (HGET key sonKey)这样是获取不到值的，加上双引号也不行
        config.setCodec(new StringCodec());
        return Redisson.create(config);
    }
}

编写Application.yml

spring:
  redis:
    host: 127.0.0.1
    port: 6379
    password: 123456
    database: 1
    connectTimeout: 3000
    timeout: 3000
    connectionPoolSize: 50
    connectionMinimumIdleSize: 50

编写启动类

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class RedissonApplication {
    public static void main(String[] args) {
        SpringApplication.run(RedissonApplication.class);

    }
}

编写测试类测试Redisson是否连接成功

import org.junit.Test;
import org.junit.runner.RunWith;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class RedissonTest {

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 插入hash
     **/
    @Test
    public void t1() {
        //先存储hash结构的库存数据
        RMap<Object, Object> skuStock = redissonClient.getMap("stockInfo");
        skuStock.put("residueStock", 10000);
        skuStock.put("quantitySold", 0);
        System.out.println("插入库存信息成功 ---  ");
    }
}

使用lua脚本实现扣减库存

代码实现

import org.junit.Test;
import org.junit.runner.RunWith;
import org.redisson.api.RMap;
import org.redisson.api.RScript;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 秒杀lua测试
 */
@RunWith(SpringRunner.class)
@SpringBootTest
public class SeckillLuaTest {

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 模拟下单操作
     **/
    @Test
    public void t1() throws InterruptedException {
        long l = System.currentTimeMillis();

        //先存储hash结构的库存数据
        RMap<Object, Object> skuStock = redissonClient.getMap("stock");
        skuStock.put("residueStock",10000); //剩余库存
        skuStock.put("quantitySold",0); //已售数量
        System.out.println("插入10000个商品数量完成 ---  ");

        //创建最大核心线程数为1000的线程池，最多同时接收10000个任务
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1000, 1000, 5, TimeUnit.SECONDS, new LinkedBlockingQueue<>(9000));
        //任务计数
        CountDownLatch countDownLatch = new CountDownLatch(10000);

        Random random = new Random();
        for (int i=0;i<10000;i++){
            threadPoolExecutor.execute(()->{
                //购买数量
                int num = random.nextInt(5);
                //扣除接口 0：库存不足下单失败  1：库存足够
                int result = Integer.valueOf(String.valueOf(deductInventory(num)));
                if(1==result){
                    System.out.println("下单成功，购买"+num+"个商品！");
                }else{
                    System.out.println("下单失败，购买"+num+"个商品但是库存不足！");
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        threadPoolExecutor.shutdown();

        System.out.println("-----------");

        System.out.println("模拟10000次下单耗时"+(System.currentTimeMillis()-l));
    }

    /*
    扣减库存lua
     */
    private Object deductInventory(int num){
		// lua脚本执行对象
        RScript script = redissonClient.getScript();
        
        // 编辑lua脚本
        StringBuilder temp = new StringBuilder();
        
        // 获取库存hash中stock -> residueStock 获取剩余库存
        // KEYS[1]相当于取keys中下标0 KEYS[2]相当于取keys中下标1
        // redis命令： HGET stock
        temp.append("local counter = redis.call('hget',KEYS[1],KEYS[2]);\n");
        
        // 当前库存减去需要购买的库存
        // ARGV[1]相当于传入的第一个num
        temp.append("local result  = counter - ARGV[1];\n");
        
        // 判断减去后是否大于等于0
        temp.append("if(result>=0 ) then\n");
        
        // 如果库存足够 将扣减后库存set到对应hash中并且使用hincrby给已售数量进行对应的增加操作
        temp.append("redis.call('hset',KEYS[1],KEYS[2],result);\n");
        temp.append("redis.call('hincrby',KEYS[1],KEYS[3],ARGV[1]);\n");
        
        // 库存信息处理成功后返回1
        temp.append("return 1;\n");
        temp.append("end;\n");
        
        // 库存不足返回0
        temp.append("return 0;\n");

		// 构建keys信息，代表hash值中所需要的key信息
        List<Object> keys = Arrays.asList("stock", "residueStock", "quantitySold");
        
        // 执行脚本
        Object result = script.eval(RScript.Mode.READ_WRITE, temp.toString(), RScript.ReturnType.VALUE,keys ,num);
        System.out.println("执行lua脚本后返回参数 "+result);
        return result;
    }
}