Redis：原理速成+项目实战——Redis实战7（优惠券秒杀+细节解决超卖、一人一单问题）

👨?🎓作者简介：一位大四、研0学生，正在努力准备大四暑假的实习
🌌上期文章：Redis：原理速成+项目实战——Redis实战6（封装缓存工具（高级写法）&&缓存总结）
📚订阅专栏：Redis速成
希望文章对你们有所帮助

这篇文章写了很久。我自己在边实现、边用jmeter来测试、边根据结果来优化我的代码，对于那些线程并发的问题，我大致是可以靠自己来解决，但是为了写好这篇文章，为了做好线程并发问题的分析，我在独立实现完之后，还是按黑马程序员的进度走了一下，他埋坑的地方其实都是线程并发问题的坑，我也自己掉一掉，并在这篇文章中进行总结。

文章中会涉及一些java面试常见的问题：常量池、Spring代理失效。如果没有印象大家可以专门找一下这些面经去了解一下。

聊到电商，一定离不开秒杀，而Redis在整个秒杀的业务中的作用是非常巨大的，接下来将会利用Redis实现全局ID，并实现秒杀，并且解决超卖问题、实现一人一单，逐渐优化业务。

全局唯一ID

每个店铺都可以发布优惠券（代金券），当用户抢购的时候，就会生成订单并且保存到tb_voucher_order这张表中：

可以发现，我们的主键ID没有使用自增长，这是因为如果使用数据库自增ID就会存在一些问题：
1、ID的规律性太明显，容易让别人猜测到信息
2、受单表数据量的限制（订单可能数据非常大，可能会分多表进行存储，但表的自增长相互之间不受影响，所以不同表之间可能会出现ID相同的情况，也就是说这种时候会违背ID的唯一性，这显然是不可以的）

而全局ID生成器，是一种分布式系统下用来生成全部唯一ID的工具，一般满足以下特性：
1、唯一性
2、高可用
3、高性能
4、递增性
5、安全性

除了第5点，Redis及其数据结构已经可以直接满足前4点的要求了，为了增加ID的安全性，不要直接使用Redis自增的数值，而是拼接一些其他信息，最终我们将ID组成定义为64位的二进制数，分别是1位符号位，31位时间戳，32位序列号：
1、符号位：1bit，永远为0
2、时间戳：31bit，以秒为单位，可以用69年
3、序列号：32bit，秒内的计数器，支持每秒产生2^32个不同的ID，这是用来处理相同秒内（时间戳相同）的多个业务
这样的结构是可以大幅度提高安全性的，不同时间下的ID一定不同，相同时间的情况下，也会因为32位的序列号而导致ID不同。

Redis实现全局唯一ID

我们在utils包下创建RedisIdWorker类：

@Component
public class RedisIdWorker {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 开始时间戳由main函数运行得到
     */
    public static final long BEGIN_TIMESTAMP = 1704499200L;
    /**
     * 序列号的位数
     */
    public static final int COUNT_BITS = 32;

    public long nextId(String keyPrefix){
        //获得当前时间
        LocalDateTime now = LocalDateTime.now();
        long nowSecond = now.toEpochSecond(ZoneOffset.UTC);
        //生成时间戳
        long timestamp = nowSecond - BEGIN_TIMESTAMP;
        /**
         * 接下来生成序列号
         * 我们的key的设置除了加上icr表示是自增长的，还需要在最后拼接一个日期字符串
         * 这是因为我们的序列号上限是2^32，并不大，如果每天的key都是一样的，这是很有可能超过上限的
         * 在后面拼接一个日期字符串，可以保证每一天的key都是不一样的，而且一天内也基本不可能到达2^32的上限
         * 这样做还有一个好处，我们以后可以根据每天或者每月来查看value值，起到统计效果
         */
        //获取当前日期，精确到天
        String date = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
        //ID自增长，这里最好用基本类型而不是包装类，因为后面还会做运算
        long count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);

        //拼接并返回，这里灵活用位运算
        return timestamp << COUNT_BITS | count;
    }

    public static void main(String[] args) {
        //定义时间为2024年1月1日00:00:00
        LocalDateTime time = LocalDateTime.of(2024, 1, 6, 0, 0, 0);
        //将时间变成变成秒数的形式
        long second = time.toEpochSecond(ZoneOffset.UTC);
        //在这里运行出来的时间作为BEGIN_TIMESETAMP
        System.out.println(second);
    }
}

编写测试代码：

    @Resource
    private RedisIdWorker redisIdWorker;

    //性能池
    private ExecutorService es = Executors.newFixedThreadPool(500);

    @Test
    void testIdWorker() throws InterruptedException {
        //因为线程池是异步的，因此我们要用CountDownLatch去截断，这样才能正常计时
        CountDownLatch latch = new CountDownLatch(300);
        Runnable task = () -> {
            for (int i = 0; i < 100; i++) {
                long id = redisIdWorker.nextId("order");
                System.out.println("id = " + id);
            }
            latch.countDown();
        };
        //将任务提交300次，并进行计时
        long begin = System.currentTimeMillis();
        for (int i = 0; i < 300; i++) {
            es.submit(task);
        }
        latch.await();//等待所有的countDown结束
        long end = System.currentTimeMillis();
        System.out.println("time = " + (end - begin));
    }

运行后可以发现，id各不重复，估计id生成的花费时间差不多只有2秒（id的打印也是会花时间的）

打开Redis客户端，可以发现我成功的生成了3万条的id：

优惠券秒杀下单

每个店铺都可以发布优惠券，分为平价券和特价券，平价券可以任意购买，而特价券需要秒杀抢购，表关系如下：
1、tb_voucher：优惠券基本信息（金额，规则等）

上面的type可以表示标识出是平价券还是特价券，如果是特价券我们也需要一些特定的信息，因此我们会专门拓展出一张表。
2、tb_seckill_voucher：优惠券库存、开始抢购时间、结束抢购时间（特价券需要此表）

添加优惠券

在VoucherController中提供一个接口，调用就可以实现添加秒杀优惠券：

虽然我们传入的参数只有Voucher，但是它也同样可以用来保存需要秒杀的券：


真正的添加不是客户来做的，要给后台来做，我们可以使用postman：

可以发现我们的数据库中已经存储了这个秒杀券：

实现秒杀下单

点击限时抢购，查看请求URL：

	说明
请求方式	POST
请求路径	voucher-order/seckill/{id}
请求参数	id,优惠券id
返回值	订单id

下单的时候我们需要判断2点：
1、秒杀是否开始或结束
2、库存是否充足

业务流程：

controller：

serviceimpl：

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 王雄俊
 * @since 2024-01-06
 */
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    //注入秒杀优惠券的service
    @Resource
    private ISeckillVoucherService seckillVoucherService;

    @Resource
    private RedisIdWorker redisIdWorker;

    @Override
    @Transactional
    public Result seckillVoucher(Long voucherId) {
        //查询优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        //判断秒杀是否开始
        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("秒杀尚未开始");
        }
        //判断秒杀是否结束
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
            return Result.fail("秒杀已经结束");
        }
        //判断库存是否充足
        if (voucher.getStock() < 1) {
            return Result.fail("库存不足");
        }
        //扣减库存，用mybatis-plus来写
        boolean success = seckillVoucherService.update().
                setSql("stock = stock - 1").
                eq("voucher_id", voucherId).update();//where条件
        if (!success){
            return Result.fail("库存不足");
        }
        System.out.println("啊啊啊啊啊");
        //创建订单，需要订单id、用户id、代金券id
        VoucherOrder voucherOrder = new VoucherOrder();
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);
        Long userId = UserHolder.getUser().getId();//用户Id去ThreadLocal中取
        voucherOrder.setUserId(userId);
        voucherOrder.setVoucherId(voucherId);
        save(voucherOrder);
        //返回订单ID
        return Result.ok(orderId);
    }
}

这边实现了最基础的订单秒杀，但是它存在很多问题

库存超卖问题

既然是秒杀，那每秒钟很可能会有成千上万的用户进行访问，那么这就对我们的并行化要求非常高，线程安全问题肯定是很重要的，上面的代码肯定是会存在线程安全问题的，我们可以用jmeter来做测试，为了方便我们到时候观察测试结果，我们去数据库手动把优惠券数量调回100，接着在jmeter中用200个线程来进行抢购：


这里设置的请求头则表示200个线程全部都由这一个用户来执行：

运行后可以看到有些请求成功，有些请求失败：

预期是有100个线程失败的，但是打开聚合报告可以发现失败的线程数量不到一半：

说明有些线程意外成功了，打开数据库，发现票数为-9，说明发生了超卖：

这会给商家带来损失。

库存超卖问题分析

假设库存容量为1（相当于一种临界资源），高并发的时候可能出现的异常情况：

也就是说，我们在某一时段会同时有多个线程查询库存的时候，得到的库存量为1，这时候都会进行扣减操作，造成超卖。
针对这种线程安全问题，常见解决方法就是直接加锁，可以分为悲观锁和乐观锁：
悲观锁：认为线程安全问题一定会发生，因此在操作数据之前先获取锁，确保线程串行执行。（Synchronized、Lock等）

乐观锁：认为线程安全问题不一定会发生，因此不加锁，只是在更新数据时去判断有没有其它线程对数据做了修改。（如果没有修改，那就是安全的；如果已经被其他线程修改说明发生了安全问题，此时可以重试或异常）

显然乐观锁的性能会好很多，但是实现起来会更复杂，我们要处理好关键的一点，那就是更新数据的时候，该如何去判断有没有其它线程对数据做了修改。
乐观锁的实现方式有2种方法（其实思想相同）：
1、版本号法：
给数据增加一个字段version，初始值为1，每次我们要修改库存量之前都需要先查询库存量与版本号，然后线程执行SQL语句，执行SQL语句必须要确定数据库中的这条数据的版本号就是查询出来的版本号，如果不相同说明有其他线程修改了数据，导致当前数据的版本号与之前查询的不一样：

2、CAS法
上面的方法加一个版本号其实是一种标识，但是我们不一定要借助version，实际上我们可以直接依靠库存量来做标识，在对数据库进行修改的时候，我们要首先判断当前数据的库存量与之前线程查询出来的库存量是否相同，不相同则说明发生线程安全问题，不能修改：

乐观锁解决超卖

我们选用CAS法来解决超卖，根据上述思想，我们只需要在SQL语句那增加一个判断库存量的条件：

测试一下上面的代码，先把数据库做还原，把订单数据删光，并还原stock为100，然后测试jmeter，可以发现jmeter中显示大量的失败，数据库中也显示没有超卖：


超卖问题确实没有出现了，但是这显然是不合常理的，200个线程抢100张票，票居然只能卖出20张。这说明乐观锁有弊端。
我们对于乐观锁的分析，是拿stock=1的情况来说的，所以当线程查询出来的stock与数据库的stock不一致的时候，足以说明票已经卖完了。
假设stock=100，当线程查询出来的stock与数据库的stock不一致的时候，并不能说明票卖完了，理论上库存量大概率不为0，该线程还是应该要能够实现买票操作，但全都因为查询的stock与数据库不一致导致有大量线程买票失败。

传统乐观锁太谨慎了！我们应该要对其进行改进！

我们不再判断查询条件，而只需要查询数据库中的stock是否大于0：

再次打开jmeter进行测试，异常率50%，解决了上述问题：

但是这不代表乐观锁就是完美的，很显然代码逻辑中要操作数据库，大量的线程就会给数据库带来压力，仅仅使用乐观锁在更高并发的场景下还是不太够的。

实现一人一单功能

我们在jmeter中的测试，200个线程全部都由一个用户来执行，因此打开订单表，我们可以发现订单全部被同一个用户买了：

商家做优惠券就是为了吸引更多的用户，一人多单可能会导致商家变相亏本。
其实思路是很简单的，我们只需要判断当前尝试抢优惠券的线程，其用户id在订单表中是否已经存在了，如果存在则不允许下单：

我们在库存修改的代码之前加上这一部分逻辑：

		//一人一单
        Long userId = UserHolder.getUser().getId();
        //查询订单
        int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        //判断是否存在
        if (count > 0){
            return Result.fail("您已购买过一次！");
        }

再次测试jmeter：

数据库显示这个用户买了10张优惠券，一人多单的问题有所缓解，但依旧存在：

这是因为上面的那一串逻辑还是存在了并发安全问题，在某一时刻还是会有很多的线程（同一个用户）进入了这部分逻辑，判断了count为0，因此进行了删减库存的操作。
这里我们肯定也要加锁，由于这一串逻辑并没有涉及到修改数据库的操作，所以我们只能加悲观锁。

	@Override
    public Result seckillVoucher(Long voucherId) {
        //查询优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        //判断秒杀是否开始
        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("秒杀尚未开始");
        }
        //判断秒杀是否结束
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
            return Result.fail("秒杀已经结束");
        }
        //判断库存是否充足
        if (voucher.getStock() < 1) {
            return Result.fail("库存不足");
        }

        //返回订单ID
        return createVoucherOrder(voucherId);
    }

    @Transactional //事务回滚放到这个函数
    public Result createVoucherOrder(Long voucherId) {
        //一人一单
        Long userId = UserHolder.getUser().getId();
        /**
         * userId值一样的，我们用同一把锁，但是每个请求一来，我们的id对象都是全新的
         * Long类型会存在这个问题，所以我们要用toString方法
         * 但是toString方法其实是对long类型new了一个字符串，所以每调用一个toString都是一个全新对象
         * 所以要加上intern()方法，从常量池中返回字符串的规范表示
         */
        synchronized (userId.toString().intern()) {
            //查询订单
            int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
            //判断是否存在
            if (count > 0) {
                return Result.fail("您已购买过一次！");
            }
            //扣减库存
            boolean success = seckillVoucherService.update().
                    setSql("stock = stock - 1").
                    eq("voucher_id", voucherId).
                    gt("stock", 0).
                    update();
            if (!success) {
                return Result.fail("库存不足");
            }
            //创建订单，需要订单id、用户id、代金券id
            VoucherOrder voucherOrder = new VoucherOrder();
            long orderId = redisIdWorker.nextId("order");
            voucherOrder.setId(orderId);
            voucherOrder.setUserId(userId);
            voucherOrder.setVoucherId(voucherId);
            save(voucherOrder);
            //返回订单ID
            return Result.ok(orderId);
        }
    }

需要注意一个细节，上面代码还是会发生并发安全问题：

我们这边的整个函数已经是被Spring托管了，所以事务的提交会在函数执行完毕之后，也就是说我们会先释放锁，再提交事务，当我们事务还没有提交完成，修改数据还没写入数据库，却又有其他线程进来了，再次发生线程并发问题。

所以，锁的范围太小了，我们应该要把整个函数都锁起来：

但，依旧有问题！直接调用createVoucherOrder方法是不行的，因为它相当于调用了this.createVoucherOrder，然而当前类并不是代理对象，这会导致Sping代理失效！
所以我们要先获得当前对象的代理对象，然后再去调用这个函数：

IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
return proxy.createVoucherOrder(voucherId);

需要引入依赖：

<dependency>
        <groupId>org.aspectj</groupId>
        <artifactId>aspectjweaver</artifactId>
</dependency>

并且在启动类中需要暴露代理对象：

运行项目，打开jmeter进行测试：


完美解决！
注意我没有在createVoucherOrder这个函数上面直接加锁，不然所有进行操作的线程都串行执行实在太影响效率了！

集群下的线程并发安全问题

现在已经通过加锁解决一人一单问题安全，但是这只能解决单机情况的，集群模式依旧不行，在这里试着模拟一下集群的方式来进行测试。
1、将服务启动2份，端口分别为8081与8082：

重启形成2个机子的集群：

2、修改nginx的conf目录下的nginx.conf文件，配置反向代理、负载均衡：

最后重新加载一下Nginx并重启：

最后访问网址，并连续刷新2次：

http://localhost:8080/api/voucher/list/1

查看后台可以发现两个启动服务都可以接受到信息，因为api（8080）包括了8081与8082，访问是以轮转的方式进行的：

这样就实现了负载均衡。

测试大家只需要在锁那里打个断点，并且在postman里面分别抢券（都用同一个用户）来进行优惠券抢购，可以发现只用1个用户信息，数据库中却少了2张券，说明又一次发生了并发问题。

从头分析一下：
1、对于一个服务中的2个线程，可能发生下面的并发问题：

2、我们解决方法是加锁：

之所以这样能实现，是因为我们锁住的对象是userId.toString().intern()，也就是从这台Tomcat常量池中取出userId.toString()，同一个userId之间肯定是相同的，因此可以锁住，防止并发。

3、但如果我们部署另外一台Tomcat，这是锁的锁监视器，其监视的内容和之前锁中的监视器内容是不一样的，那么新Tomcat的线程获取锁就会成功（获取的userId.toString()是不一样的，不理解的可以去看toString方法的源码），并成功的操作数据库，因此才会造成线程并行问题。
如下图，线程1、3发生了线程安全问题：

因此我们只能保证单个JVM下的线程安全，却无法保证集群中多个JVM的线程安全，我们需要在集群中加锁，也就是分布式锁，将在后续讲解。