【秒杀系统】零基础上手秒杀系统（一）：防止超卖

【秒杀系统】零基础上手秒杀系统（二）：令牌桶限流 + 再谈超卖

【秒杀系统】零基础上手秒杀系统（三）：抢购接口隐藏 + 单用户限制频率

【秒杀系统】零基础上手秒杀系统（四）：缓存与数据库双写问题的争议

【秒杀系统】零基础上手秒杀系统番外篇：阿里开源 MySQL 中间件 Canal 快速入门

【秒杀系统】零基础上手秒杀系统（五）：如何优雅的实现订单异步处理

【秒杀系统】从零打造秒杀系统（一）：防止超卖

前言

大家好，好久不发文章了。（快一个月了- -）最近有很多学习的新知识想和大家分享，但无奈最近项目蛮忙的，很多文章写了一半搁置在了笔记里，待以后慢慢补充发布。

本文主要是通过实际代码讲解，帮助你一步步搭建一个简易的秒杀系统。从而快速的了解秒杀系统的主要难点，并且迅速上手实际项目。

我对秒杀系统文章的规划：

从零开始打造简易秒杀系统：乐观锁防止超卖
从零开始打造简易秒杀系统：令牌桶限流
从零开始打造简易秒杀系统：Redis 缓存
从零开始打造简易秒杀系统：消息队列异步处理订单
…

欢迎关注我的公众号：后端技术漫谈（二维码见底部）

秒杀系统

秒杀系统介绍

秒杀系统相信网上已经介绍了很多了，我也不想黏贴很多定义过来了。

废话少说，秒杀系统主要应用在商品抢购的场景，比如：

电商抢购限量商品
卖周董演唱会的门票
火车票抢座
…

秒杀系统抽象来说就是以下几个步骤：

用户选定商品下单
校验库存
扣库存
创建用户订单
用户支付等后续步骤…

听起来就是个用户买商品的流程而已嘛，确实，所以我们为啥要说他是个专门的系统呢。。

为什么要做所谓的“系统”

如果你的项目流量非常小，完全不用担心有并发的购买请求，那么做这样一个系统意义不大。

但如果你的系统要像 12306 那样，接受高并发访问和下单的考验，那么你就需要一套完整的 流程保护措施 ，来保证你系统在用户流量高峰期不会被搞挂了。（就像 12306 刚开始网络售票那几年一样）

这些措施有什么呢：

严格防止超卖：库存 100 件你卖了 120 件，等着辞职吧
防止黑产：防止不怀好意的人群通过各种技术手段把你本该下发给群众的利益全收入了囊中。
保证用户体验：高并发下，别网页打不开了，支付不成功了，购物车进不去了，地址改不了了。这个问题非常之大，涉及到各种技术，也不是一下子就能讲完的，甚至根本就没法讲完。

我们先从“防止超卖”开始吧

毕竟，你网页可以卡住，最多是大家没参与到活动，上网口吐芬芳，骂你一波。但是你要是卖多了，本该拿到商品的用户可就不乐意了，轻则投诉你，重则找漏洞起诉赔偿。让你吃不了兜着走。

不能再说下去了，我这篇文章可是打着实战文章的名头，为什么我老是要讲废话啊啊啊啊啊啊。

上代码。

说好的做“简易”的秒杀系统，所以我们只用最简单的 SpringBoot 项目

建立“简易”的数据库表结构

一开始我们先来张最最最简易的结构表，参考了 crossoverjie 的秒杀系统文章。

等未来我们需要解决更多的系统问题，再扩展表结构。

一张库存表 stock，一张订单表 stock_order

```

-- Table structure for stock

DROP TABLE IF EXISTS stock ; CREATE TABLE stock ( id int(11) unsigned NOT NULL AUTO_INCREMENT, name varchar(50) NOT NULL DEFAULT '' COMMENT '名称', count int(11) NOT NULL COMMENT '库存', sale int(11) NOT NULL COMMENT '已售', version int(11) NOT NULL COMMENT '乐观锁，版本号', PRIMARY KEY ( id ) ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- Table structure for stock_order

DROP TABLE IF EXISTS stock_order ; CREATE TABLE stock_order ( id int(11) unsigned NOT NULL AUTO_INCREMENT, sid int(11) NOT NULL COMMENT '库存ID', name varchar(30) NOT NULL DEFAULT '' COMMENT '商品名称', create_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '创建时间', PRIMARY KEY ( id ) ) ENGINE=InnoDB DEFAULT CHARSET=utf8; ```

通过 HTTP 接口发起一次购买请求

代码中我们采用最传统的 Spring MVC+Mybaits 的结构

结构如下图：

Controller 层代码

提供一个 HTTP 接口: 参数为商品的 Id

@RequestMapping("/createWrongOrder/{sid}") @ResponseBody public String createWrongOrder(@PathVariable int sid) { LOGGER.info("购买物品编号sid=[{}]", sid); int id = 0; try { id = orderService.createWrongOrder(sid); LOGGER.info("创建订单id: [{}]", id); } catch (Exception e) { LOGGER.error("Exception", e); } return String.valueOf(id); }

Service 层代码

``` @Override public int createWrongOrder(int sid) throws Exception { //校验库存 Stock stock = checkStock(sid); //扣库存 saleStock(stock); //创建订单 int id = createOrder(stock); return id; }

private Stock checkStock(int sid) { Stock stock = stockService.getStockById(sid); if (stock.getSale().equals(stock.getCount())) { throw new RuntimeException("库存不足"); } return stock; }

private int saleStock(Stock stock) { stock.setSale(stock.getSale() + 1); return stockService.updateStockById(stock); }

private int createOrder(Stock stock) { StockOrder order = new StockOrder(); order.setSid(stock.getId()); order.setName(stock.getName()); int id = orderMapper.insertSelective(order); return id; } ```

发起并发购买请求

我们通过 JMeter（ https://jmeter.apache.org/ ）这个并发请求工具来模拟大量用户同时请求购买接口的场景。

注意：POSTMAN 并不支持并发请求，其请求是顺序的，而 JMeter 是多线程请求。希望以后 PostMan 能够支持吧，毕竟 JMeter 还在倔强的用 Java UI 框架。毕竟是亲儿子呢。

如何通过 JMeter 进行压力测试，请参考下文，讲的非常入门但详细，包教包会：

https://www.cnblogs.com/stulzq/p/8971531.html

我们在表里添加一个 iPhone，库存 100。（请忽略订单表里的数据，开始前我清空了）

在 JMeter 里启动 1000 个线程，无延迟同时访问接口。模拟 1000 个人，抢购 100 个产品的场景。点击启动：

你猜会卖出多少个呢，先想一想。。。

答案是：

卖出了 14 个，库存减少了 14 个，但是每个请求 Spring 都处理了，创建了 1000 个订单。

我这里该夸 Spring 强大的并发处理能力，还是该骂 MySQL 已经是个成熟的数据库，却都不会自己锁库存？

避免超卖问题：更新商品库存的版本号

为了解决上面的超卖问题，我们当然可以在 Service 层给更新表添加一个事务，这样每个线程更新请求的时候都会先去锁表的这一行（悲观锁），更新完库存后再释放锁。可这样就太慢了，1000 个线程可等不及。

我们需要乐观锁。

一个最简单的办法就是，给每个商品库存一个版本号 version 字段

我们修改代码：

Controller 层

/** * 乐观锁更新库存 * @param sid * @return */ @RequestMapping("/createOptimisticOrder/{sid}") @ResponseBody public String createOptimisticOrder(@PathVariable int sid) { int id; try { id = orderService.createOptimisticOrder(sid); LOGGER.info("购买成功，剩余库存为: [{}]", id); } catch (Exception e) { LOGGER.error("购买失败：[{}]", e.getMessage()); return "购买失败，库存不足"; } return String.format("购买成功，剩余库存为：%d", id); }

Service 层

``` @Override public int createOptimisticOrder(int sid) throws Exception { //校验库存 Stock stock = checkStock(sid); //乐观锁更新库存 saleStockOptimistic(stock); //创建订单 int id = createOrder(stock); return stock.getCount() - (stock.getSale()+1); }

private void saleStockOptimistic(Stock stock) { LOGGER.info("查询数据库，尝试更新库存"); int count = stockService.updateStockByOptimistic(stock); if (count == 0){ throw new RuntimeException("并发更新库存失败，version不匹配") ; } } ```

Mapper

<update id="updateByOptimistic" parameterType="cn.monitor4all.miaoshadao.dao.Stock"> update stock <set> sale = sale + 1, version = version + 1, </set> WHERE id = #{id,jdbcType=INTEGER} AND version = #{version,jdbcType=INTEGER} </update>

我们在实际减库存的 SQL 操作中，首先判断 version 是否是我们查询库存时候的 version，如果是，扣减库存，成功抢购。如果发现 version 变了，则不更新数据库，返回抢购失败。

发起并发购买请求

这次，我们能成功吗？

再次打开 JMeter，把库存恢复为 100，清空订单表，发起 1000 次请求。

这次的结果是：

卖出去了 39 个，version 更新为了 39,同时创建了 39 个订单。我们没有超卖，可喜可贺。

由于并发访问的原因，很多线程更新库存失败了，所以在我们这种设计下，1000 个人真要是同时发起购买，只有 39 个幸运儿能够买到东西，但是我们防止了超卖。

手速快未必好，还得看运气呀！

OK，今天先到这里，之后我们继续一步步完善这个简易的秒杀系统，它总有从树苗变成大树的那一天！

源码

我会随着文章的更新，一直同步更新项目代码，欢迎关注：

https://github.com/qqxx6661/miaosha

参考

https://cloud.tencent.com/developer/article/1488059
https://juejin.im/post/5dd09f5af265da0be72aacbd
https://crossoverjie.top/%2F2018%2F05%2F07%2Fssm%2FSSM18-seconds-kill%2F

【秒杀系统】零基础上手秒杀系统（二）：令牌桶限流 + 再谈超卖

前言

本文是秒杀系统的第二篇，通过实际代码讲解，帮助你快速的了解秒杀系统的关键点，上手实际项目。

本篇主要讲解接口限流措施，接口限流其实定义也非常广， 接口限流本身也是系统安全防护的一种措施 ，暂时列举这几种容易理解的：

令牌桶限流
单用户访问频率限流
抢购接口隐藏

此外，前文发出后很多同学对于乐观锁在高并发时无法卖出全部商品提出了“严正抗议”，所以还是在本篇中补充讲解下乐观锁与悲观锁。

前文回顾和未来文章规划：

从零开始打造简易秒杀系统：防止超卖
从零开始打造简易秒杀系统：接口限流（令牌桶限流）+ 再谈超卖
从零开始打造简易秒杀系统：接口限流（单用户限流 + 抢购接口隐藏）
从零开始打造简易秒杀系统：使用 Redis 缓存热点数据
从零开始打造简易秒杀系统：消息队列异步处理订单
…

欢迎关注我的个人公众号获取最全的原创文章：后端技术漫谈（二维码见文章底部

该项目源码在这里，妈妈再也不用担心只看文章不会实现啦：

https://github.com/qqxx6661/miaosha

秒杀系统介绍

可以翻阅该系列的第一篇文章，这里不再回顾：

从零开始搭建简易秒杀系统（一）：防止超卖

接口限流

在面临高并发的请购请求时，我们如果不对接口进行限流，可能会对后台系统造成极大的压力。尤其是对于下单的接口，过多的请求打到数据库会对系统的稳定性造成影响。

所以秒杀系统会尽量选择独立于公司其他后端系统之外进行单独部署，以免秒杀业务崩溃影响到其他系统。

除了独立部署秒杀业务之外，我们能够做的就是尽量让后台系统稳定优雅的处理大量请求。

接口限流实战：令牌桶限流算法

令牌桶限流算法网上已经有了很多介绍，我摘抄一篇介绍过来：

令牌桶算法最初来源于计算机网络。在网络传输数据时，为了防止网络拥塞，需限制流出网络的流量，使流量以比较均匀的速度向外发送。令牌桶算法就实现了这个功能，可控制发送到网络上数据的数目，并允许突发数据的发送。

大小固定的令牌桶可自行以恒定的速率源源不断地产生令牌。如果令牌不被消耗，或者被消耗的速度小于产生的速度，令牌就会不断地增多，直到把桶填满。后面再产生的令牌就会从桶中溢出。最后桶中可以保存的最大令牌数永远不会超过桶的大小。

令牌桶算法与漏桶算法

漏桶算法思路很简单，水（请求）先进入到漏桶里，漏桶以一定的速度出水，当水流入速度过大会直接溢出，可以看出漏桶算法能强行限制数据的传输速率。

令牌桶算法不能与另外一种常见算法漏桶算法相混淆。这两种算法的主要区别在于：

漏桶算法能够强行限制数据的传输速率，而令牌桶算法在能够限制数据的平均传输速率外， 还允许某种程度的突发传输 。在令牌桶算法中，只要令牌桶中存在令牌，那么就允许突发地传输数据直到达到用户配置的门限， 因此它适合于具有突发特性的流量 。

使用 Guava 的 RateLimiter 实现令牌桶限流接口

Guava 是 Google 开源的 Java 工具类，里面包罗万象，也提供了限流工具类 RateLimiter，该类里面实现了令牌桶算法。

我们拿出源码，在之前讲过的乐观锁抢购接口上增加该令牌桶限流代码：

OrderController：

``` @Controller public class OrderController {

private static final Logger LOGGER = LoggerFactory.getLogger(OrderController.class);

@Autowired
private StockService stockService;

@Autowired
private OrderService orderService;

//每秒放行10个请求
RateLimiter rateLimiter = RateLimiter.create(10);

@RequestMapping("/createWrongOrder/{sid}")
@ResponseBody
public String createWrongOrder(@PathVariable int sid) {
    int id = 0;
    try {
        id = orderService.createWrongOrder(sid);
        LOGGER.info("创建订单id: [{}]", id);
    } catch (Exception e) {
        LOGGER.error("Exception", e);
    }
    return String.valueOf(id);
}

/**
 * 乐观锁更新库存 + 令牌桶限流
 * @param sid
 * @return
 */
@RequestMapping("/createOptimisticOrder/{sid}")
@ResponseBody
public String createOptimisticOrder(@PathVariable int sid) {
    // 阻塞式获取令牌
    //LOGGER.info("等待时间" + rateLimiter.acquire());
    // 非阻塞式获取令牌
    if (!rateLimiter.tryAcquire(1000, TimeUnit.MILLISECONDS)) {
        LOGGER.warn("你被限流了，真不幸，直接返回失败");
        return "购买失败，库存不足";
    }
    int id;
    try {
        id = orderService.createOptimisticOrder(sid);
        LOGGER.info("购买成功，剩余库存为: [{}]", id);
    } catch (Exception e) {
        LOGGER.error("购买失败：[{}]", e.getMessage());
        return "购买失败，库存不足";
    }
    return String.format("购买成功，剩余库存为：%d", id);
}

} ```

代码中， RateLimiter rateLimiter = RateLimiter.create(10); 这里初始化了令牌桶类，每秒放行 10 个请求。

在接口中，可以看到有两种使用方法：

阻塞式获取令牌：请求进来后，若令牌桶里没有足够的令牌，就在这里阻塞住，等待令牌的发放。
非阻塞式获取令牌：请求进来后，若令牌桶里没有足够的令牌，会尝试等待设置好的时间（这里写了 1000ms），其会自动判断在 1000ms 后，这个请求能不能拿到令牌，如果不能拿到，直接返回抢购失败。如果 timeout 设置为 0，则等于阻塞时获取令牌。

我们使用 JMeter 设置 200 个线程，来同时抢购数据库里库存 100 个的 iPhone。（数据库结构和 JMeter 使用请查看从零开始搭建简易秒杀系统（一）：防止超卖）

我们将请求响应结果为“你被限流了，真不幸，直接返回失败”的请求单独断言出来：

我们使用 rateLimiter.tryAcquire(1000, TimeUnit.MILLISECONDS) ，非阻塞式的令牌桶算法，来看看购买结果：

可以看到， 绿色的请求代表被令牌桶拦截掉的请求 ，红色的则是购买成功下单的请求。通过 JMeter 的请求汇总报告，可以得知，在这种情况下请求能够没被限流的比率在 15% 左右。

可以看到，200 个请求中没有被限流的请求里，由于乐观锁的原因，会出现一些并发更新数据库失败的问题，导致商品没有被卖出。这也是上一篇小伙伴问的最多的问题。 所以我想再谈一谈乐观锁与悲观锁。

再谈锁之前，我们再试一试令牌桶算法的阻塞式使用，我们将代码换成 rateLimiter.acquire(); ，然后将数据库恢复成 100 个库存，订单表清零。开始请求：

这次的结果非常有意思，先放几张结果图（按顺序截图的），爱思考的同学们可以先推测下我接下来想说啥。

总结：

首先，所有请求进入了处理流程，但是被限流成每秒处理 10 个请求。
在刚开始的请求里，令牌桶里一下子被取了 10 个令牌，所以出现了第二张图中的，乐观锁并发更新失败，然而在后面的请求中，由于令牌一旦生成就被拿走，所以请求进来的很均匀，没有再出现并发更新库存的情况。 这也符合“令牌桶”的定义，可以应对突发请求（只是由于乐观锁，所以购买冲突了）。而非“漏桶”的永远恒定的请求限制。
200 个请求， 在乐观锁的情况下 ，卖出了全部 100 个商品，如果没有该限流，而请求又过于集中的话，会卖不出去几个。就像第一篇文章中的那种情况一样。

Guava 中 RateLimiter 实现原理

令牌桶的实现原理，本文中不再班门弄斧了，还是以实战为主。

毕竟 Guava 是只提供了令牌桶的一种实现，实际项目中肯定还要根据需求来使用或者自己实现，大家可以看看这篇文章：

https://segmentfault.com/a/1190000012875897

再谈防止超卖

讲完了令牌桶限流算法，我们再回头思考超卖的问题，在 海量请求 的场景下，如果像第一篇文章那样的 使用乐观锁 ，会导致大量的请求返回抢购失败，用户体验极差。

然而使用悲观锁，比如数据库事务，则可以让数据库一个个处理库存数修改，修改成功后再迎接下一个请求，所以在不同情况下，应该根据实际情况使用悲观锁和乐观锁。

悲观锁(Pessimistic Lock), 顾名思义，就是很悲观，每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会 block 直到它拿到锁。传统的关系型数据库里边就用到了很多这种锁机制，比如行锁，表锁等，读锁，写锁等，都是在做操作之前先上锁。

乐观锁(Optimistic Lock), 顾名思义，就是很乐观，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号等机制。乐观锁适用于多读的应用类型，这样可以提高吞吐量，像数据库如果提供类似于 write_condition 机制的其实都是提供的乐观锁。

两种锁各有优缺点，不能单纯的定义哪个好于哪个。

乐观锁比较适合数据修改比较少，读取比较频繁的场景，即使出现了少量的冲突，这样也省去了大量的锁的开销，故而提高了系统的吞吐量。
但是如果经常发生冲突（写数据比较多的情况下），上层应用不不断的 retry，这样反而降低了性能，对于这种情况使用悲观锁就更合适。

实现不需要版本号字段的乐观锁

上一篇文章中，我的乐观锁建立在更新数据库版本号上，这里贴出一种不用额外字段的乐观锁 SQL 语句。

<update id="updateByOptimistic" parameterType="cn.monitor4all.miaoshadao.dao.Stock"> update stock <set> sale = sale + 1, </set> WHERE id = #{id,jdbcType=INTEGER} AND sale = #{sale,jdbcType=INTEGER} </update>

实现悲观锁

我们为了在高流量下，能够更好更快的卖出商品，我们实现一个悲观锁（事务 for update 更新库存）。看看悲观锁的结果如何。

在 Controller 中，增加一个悲观锁卖商品接口：

/** * 事务for update更新库存 * @param sid * @return */ @RequestMapping("/createPessimisticOrder/{sid}") @ResponseBody public String createPessimisticOrder(@PathVariable int sid) { int id; try { id = orderService.createPessimisticOrder(sid); LOGGER.info("购买成功，剩余库存为: [{}]", id); } catch (Exception e) { LOGGER.error("购买失败：[{}]", e.getMessage()); return "购买失败，库存不足"; } return String.format("购买成功，剩余库存为：%d", id); }

在 Service 中,给该卖商品流程加上事务:

``` @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED) @Override public int createPessimisticOrder(int sid){ //校验库存(悲观锁for update) Stock stock = checkStockForUpdate(sid); //更新库存 saleStock(stock); //创建订单 int id = createOrder(stock); return stock.getCount() - (stock.getSale()); }

/* * 检查库存 ForUpdate * @param sid * @return / private Stock checkStockForUpdate(int sid) { Stock stock = stockService.getStockByIdForUpdate(sid); if (stock.getSale().equals(stock.getCount())) { throw new RuntimeException("库存不足"); } return stock; }

/* * 更新库存 * @param stock / private void saleStock(Stock stock) { stock.setSale(stock.getSale() + 1); stockService.updateStockById(stock); }

/* * 创建订单 * @param stock * @return / private int createOrder(Stock stock) { StockOrder order = new StockOrder(); order.setSid(stock.getId()); order.setName(stock.getName()); int id = orderMapper.insertSelective(order); return id; } ```

这里使用 Spring 的事务， @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED) ，如果遇到回滚，则返回 Exception，并且事务传播使用 PROPAGATION_REQUIRED–支持当前事务，如果当前没有事务，就新建一个事务 ，关于 Spring 事务传播机制可以自行查阅资料，以后也想出一个总结文章。

我们依然设置 100 个商品，清空订单表，开始用 JMeter 更改请求的接口 /createPessimisticOrder/1 ，发起 200 个请求：

查看结果，可以看到，HMeter 给出的汇总报告中，200 个请求，100 个返回了抢购成功，100 个返回了抢购失败。并且商品卖给了前 100 个进来的请求，十分的有序。

所以，悲观锁在大量请求的请求下，有着更好的卖出成功率。但是需要注意的是，如果请求量巨大，悲观锁会导致后面的请求进行了长时间的阻塞等待，用户就必须在页面等待，很像是“假死”，可以通过配合令牌桶限流，或者是给用户显著的等待提示来优化。

悲观锁真的锁住库存了吗？

最后一个问题，我想证明下我的事务真的在执行 for update 后锁住了商品库存，不让其他线程修改库存。

我们在 idea 中打断点，让代码运行到 for update 执行完成后。然后再 MySQL 命令行中，执行 update stock set count = 50 where id = 1; 试图偷偷修改库存，再回车之后，你会发现命令行阻塞了，没有返回任何消息，显然他在等待行锁的释放。

接下里，你手动继续运行程序，把该事务执行完。在事务执行完成的瞬间，命令行中成功完成了修改，说明锁已经被线程释放，其他的线程能够成功修改库存了。证明事务的行锁是有效的！

总结

本项目的代码开源在了 GitHub，大家随意使用：

https://github.com/qqxx6661/miaosha

下一篇，将会继续讲解接口限流（单用户限流 + 抢购接口隐藏）。

现在有点累，休息休息。

希望大家多多支持我的公主号：后端技术漫谈。

参考

https://cloud.tencent.com/developer/article/1488059
https://juejin.im/post/5dd09f5af265da0be72aacbd
https://crossoverjie.top/%2F2018%2F05%2F07%2Fssm%2FSSM18-seconds-kill%2F
https://www.jianshu.com/p/5d4fe4b2a726
https://segmentfault.com/a/1190000012469586
https://zhenganwen.top/posts/30bb5ce6/

【秒杀系统】零基础上手秒杀系统（三）：抢购接口隐藏 + 单用户限制频率

前言

时光飞逝，两周过去了，是时候继续填坑了，不然又要被网友喷了。

本文是秒杀系统的第三篇，通过实际代码讲解，帮助你了解秒杀系统设计的关键点，上手实际项目。

本篇主要讲解秒杀系统中，关于抢购（下单）接口相关的单用户防刷措施，主要说两块内容：

抢购接口隐藏
单用户限制频率（单位时间内限制访问次数）

当然，这两个措施放在任何系统中都有用，严格来说并不是秒杀系统独特的设计，所以今天的内容也会比较的通用。

此外，我做了一张流程图，描述了目前我们实现的秒杀接口下单流程：

前文回顾和文章规划

零基础上手秒杀系统（一）：防止超卖
零基础上手秒杀系统（二）：令牌桶限流 + 再谈超卖
零基础上手秒杀系统（三）：抢购接口隐藏 + 单用户限制频率（本篇）
零基础上手秒杀系统：使用 Redis 缓存热点数据
零基础上手秒杀系统：消息队列异步处理订单
…

欢迎关注我的个人公众号获取最全的原创文章：后端技术漫谈（二维码见文章底部）

项目源码在这里

妈妈再也不用担心只会看文章不会实现啦：

https://github.com/qqxx6661/miaosha

抢购接口隐藏

在前两篇文章的介绍下，我们完成了防止超卖商品和抢购接口的限流，已经能够防止大流量把我们的服务器直接搞炸，这篇文章中，我们要开始关心一些细节问题。

对于稍微懂点电脑的，又会动歪脑筋的人来说，点击 F12 打开浏览器的控制台，就能在点击抢购按钮后，获取我们抢购接口的链接。（手机 APP 等其他客户端可以抓包来拿到）

一旦坏蛋拿到了抢购的链接，只要稍微写点爬虫代码，模拟一个抢购请求，就可以不通过点击下单按钮，直接在代码中请求我们的接口，完成下单。所以就有了成千上万的薅羊毛军团，写一些脚本抢购各种秒杀商品。

他们只需要在抢购时刻的 000 毫秒，开始不间断发起大量请求，觉得比大家在 APP 上点抢购按钮要快，毕竟人的速度又极限，更别说 APP 说不定还要经过几层前端验证才会真正发出请求。

所以我们需要将抢购接口进行隐藏， 抢购接口隐藏（接口加盐）的具体做法 ：

每次点击秒杀按钮，先从服务器获取一个秒杀验证值（接口内判断是否到秒杀时间）。
Redis 以缓存用户 ID 和商品 ID 为 Key，秒杀地址为 Value 缓存验证值
用户请求秒杀商品的时候，要带上秒杀验证值进行校验。

大家先停下来仔细想想，通过这样的办法，能够防住通过脚本刷接口的人吗？

能，也不能。

可以防住的是直接请求接口的人，但是只要坏蛋们把脚本写复杂一点，先去请求一个验证值，再立刻请求抢购，也是能够抢购成功的。

不过坏蛋们请求验证值接口， 也需要在抢购时间开始后 ，才能请求接口拿到验证值，然后才能申请抢购接口。理论上来说在访问接口的时间上受到了限制，并且我们还能通过在验证值接口增加更复杂的逻辑，让获取验证值的接口并不快速返回验证值，进一步拉平普通用户和坏蛋们的下单时刻。所以接口加盐还是有用的！

下面我们就实现一种简单的加盐接口代码，抛砖引玉。

代码逻辑实现

代码还是使用之前的项目，我们在其上面增加两个接口：

获取验证值接口
携带验证值下单接口

由于之前我们只有两个表，一个 stock 表放库存商品，一个 stockOrder 订单表，放订购成功的记录。但是这次涉及到了用户，所以我们新增用户表，并且添加一个用户张三。并且在订单表中，不仅要记录商品 id，同时要写入用户 id。

整个 SQL 结构如下，讲究一个简洁，暂时不加入别的多余字段：

```

-- Table structure for stock

-- Records of stock

INSERT INTO stock VALUES ('1', 'iphone', '50', '0', '0'); INSERT INTO stock VALUES ('2', 'mac', '10', '0', '0');

-- Table structure for stock_order

DROP TABLE IF EXISTS stock_order ; CREATE TABLE stock_order ( id int(11) unsigned NOT NULL AUTO_INCREMENT, sid int(11) NOT NULL COMMENT '库存ID', name varchar(30) NOT NULL DEFAULT '' COMMENT '商品名称', user_id int(11) NOT NULL DEFAULT '0', create_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '创建时间', PRIMARY KEY ( id ) ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- Records of stock_order

-- Table structure for user

DROP TABLE IF EXISTS user ; CREATE TABLE user ( id bigint(20) NOT NULL AUTO_INCREMENT, user_name varchar(255) NOT NULL DEFAULT '', PRIMARY KEY ( id ) ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4;

-- Records of user

INSERT INTO user VALUES ('1', '张三'); ```

SQL 文件在开源代码里也放了，不用担心。

获取验证值接口

该接口要求传用户 id 和商品 id，返回验证值，并且该验证值

Controller 中添加方法：

/** * 获取验证值 * @return */ @RequestMapping(value = "/getVerifyHash", method = {RequestMethod.GET}) @ResponseBody public String getVerifyHash(@RequestParam(value = "sid") Integer sid, @RequestParam(value = "userId") Integer userId) { String hash; try { hash = userService.getVerifyHash(sid, userId); } catch (Exception e) { LOGGER.error("获取验证hash失败，原因：[{}]", e.getMessage()); return "获取验证hash失败"; } return String.format("请求抢购验证hash值为：%s", hash); }

UserService 中添加方法：

``` @Override public String getVerifyHash(Integer sid, Integer userId) throws Exception {

// 验证是否在抢购时间内
LOGGER.info("请自行验证是否在抢购时间内");


// 检查用户合法性
User user = userMapper.selectByPrimaryKey(userId.longValue());
if (user == null) {
    throw new Exception("用户不存在");
}
LOGGER.info("用户信息：[{}]", user.toString());

// 检查商品合法性
Stock stock = stockService.getStockById(sid);
if (stock == null) {
    throw new Exception("商品不存在");
}
LOGGER.info("商品信息：[{}]", stock.toString());

// 生成hash
String verify = SALT + sid + userId;
String verifyHash = DigestUtils.md5DigestAsHex(verify.getBytes());

// 将hash和用户商品信息存入redis
String hashKey = CacheKey.HASH_KEY.getKey() + "_" + sid + "_" + userId;
stringRedisTemplate.opsForValue().set(hashKey, verifyHash, 3600, TimeUnit.SECONDS);
LOGGER.info("Redis写入：[{}] [{}]", hashKey, verifyHash);
return verifyHash;

} ```

一个 Cache 常量枚举类 CacheKey：

``` package cn.monitor4all.miaoshadao.utils;

public enum CacheKey { HASH_KEY("miaosha_hash"), LIMIT_KEY("miaosha_limit");

private String key;

private CacheKey(String key) {
    this.key = key;
}
public String getKey() {
    return key;
}

} ```

代码解释：

可以看到在 Service 中，我们拿到用户 id 和商品 id 后，会检查商品和用户信息是否在表中存在，并且会验证现在的时间（我这里为了简化，只是写了一行 LOGGER，大家可以根据需求自行实现）。在这样的条件过滤下，才会给出 hash 值。 并且将 Hash 值写入了 Redis 中，缓存 3600 秒（1 小时），如果用户拿到这个 hash 值一小时内没下单，则需要重新获取 hash 值。

下面又到了动小脑筋的时间了，想一下，这个 hash 值，如果每次都按照商品 + 用户的信息来 md5，是不是不太安全呢。毕竟用户 id 并不一定是用户不知道的（就比如我这种用自增 id 存储的，肯定不安全），而商品 id，万一也泄露了出去，那么坏蛋们如果再知到我们是简单的 md5，那直接就把 hash 算出来了！

在代码里，我给 hash 值加了个前缀，也就是一个 salt（盐），相当于给这个固定的字符串撒了一把盐，这个盐是 HASH_KEY("miaosha_hash") ，写死在了代码里。这样黑产只要不猜到这个盐，就没办法算出来 hash 值。

这也只是一种例子，实际中，你可以把盐放在其他地方，并且不断变化，或者结合时间戳，这样就算自己的程序员也没法知道 hash 值的原本字符串是什么了。

携带验证值下单接口

用户在前台拿到了验证值后，点击下单按钮，前端携带着特征值，即可进行下单操作。

Controller 中添加方法：

/** * 要求验证的抢购接口 * @param sid * @return */ @RequestMapping(value = "/createOrderWithVerifiedUrl", method = {RequestMethod.GET}) @ResponseBody public String createOrderWithVerifiedUrl(@RequestParam(value = "sid") Integer sid, @RequestParam(value = "userId") Integer userId, @RequestParam(value = "verifyHash") String verifyHash) { int stockLeft; try { stockLeft = orderService.createVerifiedOrder(sid, userId, verifyHash); LOGGER.info("购买成功，剩余库存为: [{}]", stockLeft); } catch (Exception e) { LOGGER.error("购买失败：[{}]", e.getMessage()); return e.getMessage(); } return String.format("购买成功，剩余库存为：%d", stockLeft); }

OrderService 中添加方法：

``` @Override public int createVerifiedOrder(Integer sid, Integer userId, String verifyHash) throws Exception {

// 验证是否在抢购时间内
LOGGER.info("请自行验证是否在抢购时间内,假设此处验证成功");

// 验证hash值合法性
String hashKey = CacheKey.HASH_KEY.getKey() + "_" + sid + "_" + userId;
String verifyHashInRedis = stringRedisTemplate.opsForValue().get(hashKey);
if (!verifyHash.equals(verifyHashInRedis)) {
    throw new Exception("hash值与Redis中不符合");
}
LOGGER.info("验证hash值合法性成功");

// 检查用户合法性
User user = userMapper.selectByPrimaryKey(userId.longValue());
if (user == null) {
    throw new Exception("用户不存在");
}
LOGGER.info("用户信息验证成功：[{}]", user.toString());

// 检查商品合法性
Stock stock = stockService.getStockById(sid);
if (stock == null) {
    throw new Exception("商品不存在");
}
LOGGER.info("商品信息验证成功：[{}]", stock.toString());

//乐观锁更新库存
saleStockOptimistic(stock);
LOGGER.info("乐观锁更新库存成功");

//创建订单
createOrderWithUserInfo(stock, userId);
LOGGER.info("创建订单成功");

return stock.getCount() - (stock.getSale()+1);

} ```

代码解释：

可以看到 service 中，我们需要验证了：

商品信息
用户信息
时间
库存

如此，我们便完成了一个拥有验证的下单接口。

试验一下接口

我们先让用户 1，法外狂徒张三登场，发起请求：

http://localhost:8080/getVerifyHash?sid=1&userId=1

得到结果:

控制台输出：

别急着下单，我们看一下 Redis 里有没有存储好 key：

木偶问题，接下来，张三可以去请求下单了！

http://localhost:8080/createOrderWithVerifiedUrl?sid=1&userId=1&verifyHash=d4ff4c458da98f69b880dd79c8a30bcf

得到输出结果：

法外狂徒张三抢购成功了！

单用户限制频率

假设我们做好了接口隐藏，但是像我上面说的，总有无聊的人会写一个复杂的脚本，先请求 hash 值，再立刻请求购买，如果你的 app 下单按钮做的很差，大家都要开抢后 0.5 秒才能请求成功，那可能会让脚本依然能够在大家前面抢购成功。

我们需要在做一个额外的措施，来限制单个用户的抢购频率。

其实很简单的就能想到用 Redis 给每个用户做访问统计，甚至是带上商品 id，对单个商品做访问统计，这都是可行的。

我们先实现一个对用户的访问频率限制，我们在用户申请下单时，检查用户的访问次数，超过访问次数，则不让他下单！

使用 Redis/Memcached

我们使用外部缓存来解决问题，这样即便是分布式的秒杀系统，请求被随意分流的情况下，也能做到精准的控制每个用户的访问次数。

Controller 中添加方法：

/** * 要求验证的抢购接口 + 单用户限制访问频率 * @param sid * @return */ @RequestMapping(value = "/createOrderWithVerifiedUrlAndLimit", method = {RequestMethod.GET}) @ResponseBody public String createOrderWithVerifiedUrlAndLimit(@RequestParam(value = "sid") Integer sid, @RequestParam(value = "userId") Integer userId, @RequestParam(value = "verifyHash") String verifyHash) { int stockLeft; try { int count = userService.addUserCount(userId); LOGGER.info("用户截至该次的访问次数为: [{}]", count); boolean isBanned = userService.getUserIsBanned(userId); if (isBanned) { return "购买失败，超过频率限制"; } stockLeft = orderService.createVerifiedOrder(sid, userId, verifyHash); LOGGER.info("购买成功，剩余库存为: [{}]", stockLeft); } catch (Exception e) { LOGGER.error("购买失败：[{}]", e.getMessage()); return e.getMessage(); } return String.format("购买成功，剩余库存为：%d", stockLeft); }

UserService 中增加两个方法：

addUserCount：每当访问订单接口，则增加一次访问次数，写入 Redis
getUserIsBanned：从 Redis 读出该用户的访问次数，超过 10 次则不让购买了！不能让张三做法外狂徒。

``` @Override public int addUserCount(Integer userId) throws Exception { String limitKey = CacheKey.LIMIT_KEY.getKey() + "_" + userId; String limitNum = stringRedisTemplate.opsForValue().get(limitKey); int limit = -1; if (limitNum == null) { stringRedisTemplate.opsForValue().set(limitKey, "0", 3600, TimeUnit.SECONDS); } else { limit = Integer.parseInt(limitNum) + 1; stringRedisTemplate.opsForValue().set(limitKey, String.valueOf(limit), 3600, TimeUnit.SECONDS); } return limit; }

@Override
public boolean getUserIsBanned(Integer userId) {
    String limitKey = CacheKey.LIMIT_KEY.getKey() + "_" + userId;
    String limitNum = stringRedisTemplate.opsForValue().get(limitKey);
    if (limitNum == null) {
        LOGGER.error("该用户没有访问申请验证值记录，疑似异常");
        return true;
    }
    return Integer.parseInt(limitNum) > ALLOW_COUNT;
}

```

试一试接口

使用前文用的 JMeter 做并发访问接口 30 次，可以看到下单了 10 次后，不让再购买了：

大功告成了。

能否不用 Redis/Memcached 实现用户访问频率统计

且慢， 如果你说你不愿意用 Redis ，有什么办法能够实现访问频率统计吗，有呀，如果你放弃分布式的部署服务，那么你可以在内存中存储访问次数，比如：

Google Guava 的内存缓存
状态模式

不知道大家的设计模式复习的怎么样了，如果没有复习到状态模式，可以先去看看状态模式的定义。状态模式很适合实现这种访问次数限制场景。

我的博客和公众号（后端技术漫谈）里，写了个《设计模式自习室》系列，详细介绍了每种设计模式，大家有兴趣可可以看看。【设计模式自习室】开篇：为什么要有设计模式？

这里我就不实现了，毕竟咱们还是分布式秒杀服务为主，不过引用一个博客的例子，大家感受下状态模式的实际应用：

https://www.cnblogs.com/java-my-life/archive/2012/06/08/2538146.html

考虑一个在线投票系统的应用，要实现控制同一个用户只能投一票，如果一个用户反复投票，而且投票次数超过 5 次，则判定为恶意刷票，要取消该用户投票的资格，当然同时也要取消他所投的票；如果一个用户的投票次数超过 8 次，将进入黑名单，禁止再登录和使用系统。

``` public class VoteManager { //持有状体处理对象 private VoteState state = null; //记录用户投票的结果，Map 对应Map<用户名称，投票的选项> private Map mapVote = new HashMap (); //记录用户投票次数，Map 对应Map<用户名称，投票的次数> private Map mapVoteCount = new HashMap (); / * 获取用户投票结果的Map */ public Map getMapVote() { return mapVote; } / * 投票 * @param user 投票人 * @param voteItem 投票的选项 */ public void vote(String user,String voteItem){ //1.为该用户增加投票次数 //从记录中取出该用户已有的投票次数 Integer oldVoteCount = mapVoteCount.get(user); if(oldVoteCount == null){ oldVoteCount = 0; } oldVoteCount += 1; mapVoteCount.put(user, oldVoteCount); //2.判断该用户的投票类型，就相当于判断对应的状态 //到底是正常投票、重复投票、恶意投票还是上黑名单的状态 if(oldVoteCount == 1){ state = new NormalVoteState(); } else if(oldVoteCount > 1 && oldVoteCount < 5){ state = new RepeatVoteState(); } else if(oldVoteCount >= 5 && oldVoteCount <8){ state = new SpiteVoteState(); } else if(oldVoteCount > 8){ state = new BlackVoteState(); } //然后转调状态对象来进行相应的操作 state.vote(user, voteItem, this); } } public class Client {

public static void main(String[] args) {

    VoteManager vm = new VoteManager();
    for(int i=0;i<9;i++){
        vm.vote("u1","A");
    }
}

} ```

结果：

总结

本项目的代码开源在了 GitHub，大家随意使用：

https://github.com/qqxx6661/miaosha

最后，感谢大家的喜爱。

希望大家多多支持我的公主号：后端技术漫谈。

参考

https://cloud.tencent.com/developer/article/1488059
https://juejin.im/post/5dd09f5af265da0be72aacbd
https://zhenganwen.top/posts/30bb5ce6/
https://www.cnblogs.com/java-my-life/archive/2012/06/08/2538146.html

秒杀系统实战（四）| 缓存与数据库双写问题的争议

前言

微笑挖坑，努力填坑。

———— 已经拥有黑眼圈，但还没学会小猪老师时间管理学的蛮三刀同学。

本文是秒杀系统的第四篇，我们来讨论秒杀系统中 「缓存热点数据」 的问题，进一步延伸到数据库和缓存的双写一致性问题，并且给出了实现代码。

前文回顾和文章规划

零基础上手秒杀系统（一）：防止超卖
零基础上手秒杀系统（二）：令牌桶限流 + 再谈超卖
零基础上手秒杀系统（三）：抢购接口隐藏 + 单用户限制频率
零基础上手秒杀系统（四）：缓存与数据库双写问题的争议（本篇）
零基础上手秒杀系统：消息队列异步处理订单
...

本篇文章主要内容

缓存热点数据
- 为何要使用缓存
哪类数据适合缓存
缓存的利与弊
缓存和数据库双写一致性
- 不使用更新缓存而是删除缓存
先删除缓存，还是先操作数据库？
我一定要数据库和缓存数据一致怎么办
实战：先删除缓存，再更新数据库
实战：先更新数据库，再删缓存
实战：删除缓存重试机制
实战：删除缓存重试机制
实战：读取 binlog 异步删除缓存

项目源码在这里

妈妈再也不用担心我看完文章不会写代码实现啦：

https://github.com/qqxx6661/miaosha

正文

缓存热点数据

在秒杀实际的业务中，一定有很多需要做缓存的场景，比如售卖的商品，包括名称，详情等。访问量很大的数据，可以算是“热点”数据了，尤其是一些读取量远大于写入量的数据，更应该被缓存，而不应该让请求打到数据库上。

为何要使用缓存

缓存是为了追求“快”而存在的。我们用代码举一个例子。

拿出我之前三篇文章的项目代码来，在其中增加两个查询库存的接口 getStockByDB 和 getStockByCache，分别表示从数据库和缓存查询某商品的库存量。

随后我们用 JMeter 进行并发请求测试。（JMeter 的使用请参考我的第一篇秒杀系统文章）

``` /* * 查询库存：通过数据库查询库存 * @param sid * @return / @RequestMapping("/getStockByDB/{sid}") @ResponseBody public String getStockByDB(@PathVariable int sid) { int count; try { count = stockService.getStockCountByDB(sid); } catch (Exception e) { LOGGER.error("查询库存失败：[{}]", e.getMessage()); return "查询库存失败"; } LOGGER.info("商品Id: [{}] 剩余库存为: [{}]", sid, count); return String.format("商品Id: %d 剩余库存为：%d", sid, count); }

/* * 查询库存：通过缓存查询库存 * 缓存命中：返回库存 * 缓存未命中：查询数据库写入缓存并返回 * @param sid * @return / @RequestMapping("/getStockByCache/{sid}") @ResponseBody public String getStockByCache(@PathVariable int sid) { Integer count; try { count = stockService.getStockCountByCache(sid); if (count == null) { count = stockService.getStockCountByDB(sid); LOGGER.info("缓存未命中，查询数据库，并写入缓存"); stockService.setStockCountToCache(sid, count); } } catch (Exception e) { LOGGER.error("查询库存失败：[{}]", e.getMessage()); return "查询库存失败"; } LOGGER.info("商品Id: [{}] 剩余库存为: [{}]", sid, count); return String.format("商品Id: %d 剩余库存为：%d", sid, count); } ```

在设置为 10000 个并发请求的情况下，运行 JMeter，结果首先出现了大量的报错，10000 个请求中 98% 的请求都直接失败了。打开日志，报错如下：

原来是 SpringBoot 内置的 Tomcat 最大并发数搞的鬼，其默认值为 200，对于 10000 的并发，单机服务实在是力不从心。当然，你可以修改这里的并发数设置，但是你的小机器仍然可能会扛不住。

将其修改为如下配置后，我的小机器才在通过缓存拿库存的情况下，保证了 10000 个并发的 100% 返回请求：

server.tomcat.max-threads=10000 server.tomcat.max-connections=10000

不使用缓存的情况下，吞吐量为 668 个请求每秒，并且有 5% 的请求由于服务压力实在太大，没有返回库存数据：

使用缓存的情况下，吞吐量为 2177 个请求每秒：

在这种“不严谨”的对比下，有缓存对于一台单机，性能提升了 3 倍多，如果在多台机器，更多并发的情况下，由于数据库有了更大的压力，缓存的性能优势应该会更加明显。

测完了这个小实验，我看了眼我挂着 MySQL 的小水管腾讯云服务器，生怕他被这么高流量搞挂。这种突发的流量，指不定会被检测为异常攻击流量呢~

我用的是腾讯云服务器 1C4G2M，活动买的，很便宜。

哪类数据适合缓存

缓存量大但又不常变化的数据，比如详情，评论等。对于那些经常变化的数据，其实并不适合缓存，一方面会增加系统的复杂性（缓存的更新，缓存脏数据），另一方面也给系统带来一定的不稳定性（缓存系统的维护）。

「但一些极端情况下，你需要将一些会变动的数据进行缓存，比如想要页面显示准实时的库存数，或者其他一些特殊业务场景。这时候你需要保证缓存不能（一直）有脏数据，这就需要再深入讨论一下。」

缓存的利与弊

我们到底该不该上缓存的，这其实也是个 trade-off 的问题。

上缓存的优点：

能够缩短服务的响应时间，给用户带来更好的体验。
能够增大系统的吞吐量，依然能够提升用户体验。
减轻数据库的压力，防止高峰期数据库被压垮，导致整个线上服务 BOOM！

上了缓存，也会引入很多额外的问题：

缓存有多种选型，是内存缓存，Memcached 还是 Redis，你是否都熟悉，如果不熟悉，无疑增加了维护的难度（本来是个纯洁的数据库系统）。
缓存系统也要考虑分布式，比如 Redis 的分布式缓存还会有很多坑，无疑增加了系统的复杂性。
在特殊场景下，如果对缓存的准确性有非常高的要求，就必须考虑 「缓存和数据库的一致性问题」 。

「本文想要重点讨论的，就是缓存和数据库的一致性问题，客观且往下看。」

缓存和数据库双写一致性

说了这么多缓存的必要性，那么使用缓存是不是就是一个很简单的事情了呢，我之前也一直是这么觉得的，直到遇到了需要缓存与数据库保持强一致的场景，才知道让数据库数据和缓存数据保持一致性是一门很高深的学问。

从远古的硬件缓存，操作系统缓存开始，缓存就是一门独特的学问。这个问题也被业界探讨了非常久，争论至今。我翻阅了很多资料，发现其实这是一个权衡的问题。值得好好讲讲。

以下的讨论会引入几方观点，我会跟着观点来写代码验证所提到的问题。

不使用更新缓存而是删除缓存

「大部分观点认为，做缓存不应该是去更新缓存，而是应该删除缓存，然后由下个请求去去缓存，发现不存在后再读取数据库，写入缓存。」

《分布式之数据库和缓存双写一致性方案解析》孤独烟：

「原因一：线程安全角度」

同时有请求 A 和请求 B 进行更新操作，那么会出现

（1）线程 A 更新了数据库

（2）线程 B 更新了数据库

（3）线程 B 更新了缓存

（4）线程 A 更新了缓存

这就出现请求 A 更新缓存应该比请求 B 更新缓存早才对，但是因为网络等原因，B 却比 A 更早更新了缓存。这就导致了脏数据，因此不考虑。

「原因二：业务场景角度」

有如下两点：

（1）如果你是一个写数据库场景比较多，而读数据场景比较少的业务需求，采用这种方案就会导致，数据压根还没读到，缓存就被频繁的更新，浪费性能。

（2）如果你写入数据库的值，并不是直接写入缓存的，而是要经过一系列复杂的计算再写入缓存。那么，每次写入数据库后，都再次计算写入缓存的值，无疑是浪费性能的。显然，删除缓存更为适合。

「其实如果业务非常简单，只是去数据库拿一个值，写入缓存，那么更新缓存也是可以的。但是，淘汰缓存操作简单，并且带来的副作用只是增加了一次 cache miss，建议作为通用的处理方式。」

先删除缓存，还是先操作数据库？

「那么问题就来了，我们是先删除缓存，然后再更新数据库，还是先更新数据库，再删缓存呢？」

先来看看大佬们怎么说。

《【58 沈剑架构系列】缓存架构设计细节二三事》58 沈剑：

对于一个不能保证事务性的操作，一定涉及“哪个任务先做，哪个任务后做”的问题，解决这个问题的方向是：如果出现不一致，谁先做对业务的影响较小，就谁先执行。

假设先淘汰缓存，再写数据库：第一步淘汰缓存成功，第二步写数据库失败，则只会引发一次 Cache miss。

假设先写数据库，再淘汰缓存：第一步写数据库操作成功，第二步淘汰缓存失败，则会出现 DB 中是新数据，Cache 中是旧数据，数据不一致。

沈剑老师说的没有问题，不过 「没完全考虑好并发请求时的数据脏读问题」 ，让我们再来看看孤独烟老师《分布式之数据库和缓存双写一致性方案解析》：

「先删缓存，再更新数据库」

该方案会导致请求数据不一致

同时有一个请求 A 进行更新操作，另一个请求 B 进行查询操作。那么会出现如下情形:

（1）请求 A 进行写操作，删除缓存

（2）请求 B 查询发现缓存不存在

（3）请求 B 去数据库查询得到旧值

（4）请求 B 将旧值写入缓存

（5）请求 A 将新值写入数据库

上述情况就会导致不一致的情形出现。而且，如果不采用给缓存设置过期时间策略，该数据永远都是脏数据。

「所以先删缓存，再更新数据库并不是一劳永逸的解决方案，再看看先更新数据库，再删缓存」

「先更新数据库，再删缓存」 这种情况不存在并发问题么？

不是的。假设这会有两个请求，一个请求 A 做查询操作，一个请求 B 做更新操作，那么会有如下情形产生

（1）缓存刚好失效

（2）请求 A 查询数据库，得一个旧值

（3）请求 B 将新值写入数据库

（4）请求 B 删除缓存

（5）请求 A 将查到的旧值写入缓存

ok，如果发生上述情况，确实是会发生脏数据。

然而，发生这种情况的概率又有多少呢？

发生上述情况有一个先天性条件，就是步骤（3）的写数据库操作比步骤（2）的读数据库操作耗时更短，才有可能使得步骤（4）先于步骤（5）。可是，大家想想，「数据库的读操作的速度远快于写操作的（不然做读写分离干嘛，做读写分离的意义就是因为读操作比较快，耗资源少），因此步骤（3）耗时比步骤（2）更短，这一情形很难出现。」

「先更新数据库，再删缓存」 依然会有问题，不过，问题出现的可能性会因为上面说的原因，变得比较低！

所以，如果你想实现基础的缓存数据库双写一致的逻辑，那么在大多数情况下，在不想做过多设计，增加太大工作量的情况下，请 「先更新数据库，再删缓存!」

我一定要数据库和缓存数据一致怎么办

那么，如果我 tm 非要保证绝对一致性怎么办，先给出结论：

「没有办法做到绝对的一致性，这是由 CAP 理论决定的，缓存系统适用的场景就是非强一致性的场景，所以它属于 CAP 中的 AP。」

所以，我们得委曲求全，可以去做到 BASE 理论中说的 「最终一致性」 。

最终一致性强调的是系统中所有的数据副本，在经过一段时间的同步后，最终能够达到一个一致的状态。因此，最终一致性的本质是需要系统保证最终数据能够达到一致，而不需要实时保证系统数据的强一致性

大佬们给出了到达最终一致性的解决思路，主要是针对上面两种双写策略（先删缓存，再更新数据库/先更新数据库，再删缓存）导致的脏数据问题，进行相应的处理，来保证最终一致性。

延时双删

问：先删除缓存，再更新数据库中避免脏数据？

答案：采用延时双删策略。

上文我们提到，在先删除缓存，再更新数据库的情况下，如果不采用给缓存设置过期时间策略，该数据永远都是脏数据。

「那么延时双删怎么解决这个问题呢？」

（1）先淘汰缓存

（2）再写数据库（这两步和原来一样）

（3）休眠 1 秒，再次淘汰缓存

这么做，可以将 1 秒内所造成的缓存脏数据，再次删除。

「那么，这个 1 秒怎么确定的，具体该休眠多久呢？」

针对上面的情形，读者应该自行评估自己的项目的读数据业务逻辑的耗时。然后写数据的休眠时间则在读数据业务逻辑的耗时基础上，加几百 ms 即可。这么做的目的，就是确保读请求结束，写请求可以删除读请求造成的缓存脏数据。

「如果你用了 MySQL 的读写分离架构怎么办？」

ok，在这种情况下，造成数据不一致的原因如下，还是两个请求，一个请求 A 进行更新操作，另一个请求 B 进行查询操作。

（1）请求 A 进行写操作，删除缓存

（2）请求 A 将数据写入数据库了，

（3）请求 B 查询缓存发现，缓存没有值

（4）请求 B 去从库查询，这时，还没有完成主从同步，因此查询到的是旧值

（5）请求 B 将旧值写入缓存

（6）数据库完成主从同步，从库变为新值

上述情形，就是数据不一致的原因。还是使用双删延时策略。只是，睡眠时间修改为在主从同步的延时时间基础上，加几百 ms。

「采用这种同步淘汰策略，吞吐量降低怎么办？」

ok，那就将第二次删除作为异步的。自己起一个线程，异步删除。这样，写的请求就不用沉睡一段时间后了，再返回。这么做，加大吞吐量。

「所以在先删除缓存，再更新数据库的情况下」 ，可以使用延时双删的策略，来保证脏数据只会存活一段时间，就会被准确的数据覆盖。

「在先更新数据库，再删缓存的情况下」 ，缓存出现脏数据的情况虽然可能性极小，但也会出现。我们依然可以用延时双删策略，在请求 A 对缓存写入了脏的旧值之后，再次删除缓存。来保证去掉脏缓存。

删缓存失败了怎么办：重试机制

看似问题都已经解决了，但其实，还有一个问题没有考虑到，那就是删除缓存的操作，失败了怎么办？比如延时双删的时候，第二次缓存删除失败了，那不还是没有清除脏数据吗？

「解决方案就是再加上一个重试机制，保证删除缓存成功。」

参考孤独烟老师给的方案图：

「方案一：」

流程如下所示

（1）更新数据库数据；

（2）缓存因为种种问题删除失败

（3）将需要删除的 key 发送至消息队列

（4）自己消费消息，获得需要删除的 key

（5）继续重试删除操作，直到成功

然而，该方案有一个缺点，对业务线代码造成大量的侵入。于是有了方案二，在方案二中，启动一个订阅程序去订阅数据库的 binlog，获得需要操作的数据。在应用程序中，另起一段程序，获得这个订阅程序传来的信息，进行删除缓存操作。

方案二：

流程如下图所示：

（1）更新数据库数据

（2）数据库会将操作信息写入 binlog 日志当中

（3）订阅程序提取出所需要的数据以及 key

（4）另起一段非业务代码，获得该信息

（5）尝试删除缓存操作，发现删除失败

（6）将这些信息发送至消息队列

（7）重新从消息队列中获得该数据，重试操作。

「而读取 binlog 的中间件，可以采用阿里开源的 canal」

好了，到这里我们已经把缓存双写一致性的思路彻底梳理了一遍，下面就是我对这几种思路徒手写的实战代码，方便有需要的朋友参考。

实战：先删除缓存，再更新数据库

终于到了实战，我们在秒杀项目的代码上增加接口：先删除缓存，再更新数据库

OrderController 中新增：

/** * 下单接口：先删除缓存，再更新数据库 * @param sid * @return */ @RequestMapping("/createOrderWithCacheV1/{sid}") @ResponseBody public String createOrderWithCacheV1(@PathVariable int sid) { int count = 0; try { // 删除库存缓存 stockService.delStockCountCache(sid); // 完成扣库存下单事务 orderService.createPessimisticOrder(sid); } catch (Exception e) { LOGGER.error("购买失败：[{}]", e.getMessage()); return "购买失败，库存不足"; } LOGGER.info("购买成功，剩余库存为: [{}]", count); return String.format("购买成功，剩余库存为：%d", count); }

stockService 中新增：

@Override public void delStockCountCache(int id) { String hashKey = CacheKey.STOCK_COUNT.getKey() + "_" + id; stringRedisTemplate.delete(hashKey); LOGGER.info("删除商品id：[{}] 缓存", id); }

其他涉及的代码都在之前三篇文章中有介绍，并且可以直接去 GitHub 拿到项目源码，就不在这里重复贴了。

实战：先更新数据库，再删缓存

如果是先更新数据库，再删缓存，那么代码只是在业务顺序上颠倒了一下，这里就只贴 OrderController 中新增：

/** * 下单接口：先更新数据库，再删缓存 * @param sid * @return */ @RequestMapping("/createOrderWithCacheV2/{sid}") @ResponseBody public String createOrderWithCacheV2(@PathVariable int sid) { int count = 0; try { // 完成扣库存下单事务 orderService.createPessimisticOrder(sid); // 删除库存缓存 stockService.delStockCountCache(sid); } catch (Exception e) { LOGGER.error("购买失败：[{}]", e.getMessage()); return "购买失败，库存不足"; } LOGGER.info("购买成功，剩余库存为: [{}]", count); return String.format("购买成功，剩余库存为：%d", count); }

实战：缓存延时双删

如何做延时双删呢，最好的方法是开设一个线程池，在线程中删除 key，而不是使用 Thread.sleep 进行等待，这样会阻塞用户的请求。

更新前先删除缓存，然后更新数据，再延时删除缓存。

OrderController 中新增接口：

``` // 延时时间：预估读数据库数据业务逻辑的耗时，用来做缓存再删除 private static final int DELAY_MILLSECONDS = 1000;

/* * 下单接口：先删除缓存，再更新数据库，缓存延时双删 * @param sid * @return / @RequestMapping("/createOrderWithCacheV3/{sid}") @ResponseBody public String createOrderWithCacheV3(@PathVariable int sid) { int count; try { // 删除库存缓存 stockService.delStockCountCache(sid); // 完成扣库存下单事务 count = orderService.createPessimisticOrder(sid); // 延时指定时间后再次删除缓存 cachedThreadPool.execute(new delCacheByThread(sid)); } catch (Exception e) { LOGGER.error("购买失败：[{}]", e.getMessage()); return "购买失败，库存不足"; } LOGGER.info("购买成功，剩余库存为: [{}]", count); return String.format("购买成功，剩余库存为：%d", count); } ```

OrderController 中新增线程池：

``` // 延时双删线程池 private static ExecutorService cachedThreadPool = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,new SynchronousQueue ());

/* * 缓存再删除线程 / private class delCacheByThread implements Runnable { private int sid; public delCacheByThread(int sid) { this.sid = sid; } public void run() { try { LOGGER.info("异步执行缓存再删除，商品id：[{}]，首先休眠：[{}] 毫秒", sid, DELAY_MILLSECONDS); Thread.sleep(DELAY_MILLSECONDS); stockService.delStockCountCache(sid); LOGGER.info("再次删除商品id：[{}] 缓存", sid); } catch (Exception e) { LOGGER.error("delCacheByThread执行出错", e); } } } ```

来试验一下，请求接口 createOrderWithCacheV3：

日志中，做到了两次删除：

实战：删除缓存重试机制

上文提到了，要解决删除失败的问题，需要用到消息队列，进行删除操作的重试。这里我们为了达到效果，接入了 RabbitMq，并且需要在接口中写发送消息，并且需要消费者常驻来消费消息。Spring 整合 RabbitMq 还是比较简单的，我把简单的整合代码也贴出来。

pom.xml 新增 RabbitMq 的依赖：

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

写一个 RabbitMqConfig：

``` @Configuration public class RabbitMqConfig {

@Bean
public Queue delCacheQueue() {
    return new Queue("delCache");
}

} ```

添加一个消费者：

``` @Component @RabbitListener(queues = "delCache") public class DelCacheReceiver {

private static final Logger LOGGER = LoggerFactory.getLogger(DelCacheReceiver.class);

@Autowired
private StockService stockService;

@RabbitHandler
public void process(String message) {
    LOGGER.info("DelCacheReceiver收到消息: " + message);
    LOGGER.info("DelCacheReceiver开始删除缓存: " + message);
    stockService.delStockCountCache(Integer.parseInt(message));
}

} ```

OrderController 中新增接口：

``` /* * 下单接口：先更新数据库，再删缓存，删除缓存重试机制 * @param sid * @return / @RequestMapping("/createOrderWithCacheV4/{sid}") @ResponseBody public String createOrderWithCacheV4(@PathVariable int sid) { int count; try { // 完成扣库存下单事务 count = orderService.createPessimisticOrder(sid); // 删除库存缓存 stockService.delStockCountCache(sid); // 延时指定时间后再次删除缓存 // cachedThreadPool.execute(new delCacheByThread(sid)); // 假设上述再次删除缓存没成功，通知消息队列进行删除缓存 sendDelCache(String.valueOf(sid));

} catch (Exception e) {
    LOGGER.error("购买失败：[{}]", e.getMessage());
    return "购买失败，库存不足";
}
LOGGER.info("购买成功，剩余库存为: [{}]", count);
return String.format("购买成功，剩余库存为：%d", count);

} ```

访问 createOrderWithCacheV4：

可以看到，我们先完成了下单，然后删除了缓存，并且假设延迟删除缓存失败了，发送给消息队列重试的消息，消息队列收到消息后再去删除缓存。

实战：读取 binlog 异步删除缓存

我们需要用到阿里开源的 canal 来读取 binlog 进行缓存的异步删除。

不过很蛋疼的是，这次文章的工作量实在有点太大了，连续写代码和整理文字身体有点吃不消了，不知道你们有没有学累。我准备把 canal 之后单开一个文章写，可能就是第五篇，也可能是单开一个 canal 的文章。我得先休息会，这里就留一个坑把。

扩展阅读

更新缓存的的 Design Pattern 有四种：Cache aside, Read through, Write through, Write behind caching，这里有陈皓的总结文章可以进行学习。

https://coolshell.cn/articles/17416.html

小结

引用陈浩《缓存更新的套路》最后的总结语作为小结：

分布式系统里要么通过 2PC 或是 Paxos 协议保证一致性，要么就是拼命的降低并发时脏数据的概率

缓存系统适用的场景就是非强一致性的场景，所以它属于 CAP 中的 AP，BASE 理论。

异构数据库本来就没办法强一致， 「只是尽可能减少时间窗口，达到最终一致性」 。

还有别忘了设置过期时间，这是个兜底方案

结束语

本文总结了秒杀系统中关于缓存数据的思考和实现，并探讨了缓存数据库双写一致性问题。

「可以总结为如下几点：」

对于读多写少的数据，请使用缓存。
为了保持一致性，会导致系统吞吐量的下降。
为了保持一致性，会导致业务代码逻辑复杂。
缓存做不到绝对一致性，但可以做到最终一致性。
对于需要保证缓存数据库数据一致的情况，请尽量考虑对一致性到底有多高要求，选定合适的方案，避免过度设计。

作者水平有限，写文章过程中难免出现错误和疏漏，请理性讨论与指正。

希望大家多多支持我的公主号：后端技术漫谈

参考

秒杀系统实战（五）| 如何优雅的实现订单异步处理

前言

我回来啦，前段时间忙得不可开交。这段时间终于能喘口气了，继续把之前挖的坑填起来。写完上一篇秒杀系统（四）：数据库与缓存双写一致性深入分析后，感觉文章深度一下子被我抬高了一些，现在构思新文章的时候，反而畏手畏脚，不敢随便写了。对于未来文章内容的想法，我写在了本文的末尾。

本文我们来聊聊秒杀系统中的订单异步处理。

本篇文章主要内容

为何我们需要对下订单采用异步处理
简单的订单异步处理实现
非异步与异步下单接口的性能对比
一个用户抢购体验更好的实现方式

前文回顾

项目源码

再也不用担心看完文章不会代码实现啦：

https://github.com/qqxx6661/miaosha

我发现该仓库的 star 数不知不觉已经超过 100 啦。

我努力将整个仓库的代码尽量做到整洁和可复用，在代码中我尽量做好每个方法的文档，并且尽量最小化方法的功能，比如下面这样：

``` public interface StockService { /* * 查询库存：通过缓存查询库存 * 缓存命中：返回库存 * 缓存未命中：查询数据库写入缓存并返回 * @param id * @return / Integer getStockCount(int id);

/**
 * 获取剩余库存：查数据库
 * @param id
 * @return
 */
int getStockCountByDB(int id);

/**
 * 获取剩余库存: 查缓存
 * @param id
 * @return
 */
Integer getStockCountByCache(int id);

/**
 * 将库存插入缓存
 * @param id
 * @return
 */
void setStockCountCache(int id, int count);

/**
 * 删除库存缓存
 * @param id
 */
void delStockCountCache(int id);

/**
 * 根据库存 ID 查询数据库库存信息
 * @param id
 * @return
 */
Stock getStockById(int id);

/**
 * 根据库存 ID 查询数据库库存信息（悲观锁）
 * @param id
 * @return
 */
Stock getStockByIdForUpdate(int id);

/**
 * 更新数据库库存信息
 * @param stock
 * return
 */
int updateStockById(Stock stock);

/**
 * 更新数据库库存信息（乐观锁）
 * @param stock
 * @return
 */
public int updateStockByOptimistic(Stock stock);

} ```

「这样就像一个可拔插（plug-in）模块一样，尽量让小伙伴们可以复制粘贴，整合到自己的代码里，稍作修改适配便可以使用。」

正文

秒杀系统介绍

可以翻阅该系列的第一篇文章，这里不再回顾：

零基础实现秒杀系统（一）：防止超卖

简单的订单异步处理实现

介绍

前面几篇文章，我们从 「限流角度，缓存角度」 来优化了用户下单的速度，减少了服务器和数据库的压力。这些处理对于一个秒杀系统都是非常重要的，并且效果立竿见影，那还有什么操作也能有立竿见影的效果呢？答案是对于下单的异步处理。

在秒杀系统用户进行抢购的过程中，由于在同一时间会有大量请求涌入服务器，如果每个请求都立即访问数据库进行扣减库存 + 写入订单的操作，对数据库的压力是巨大的。

如何减轻数据库的压力呢，「我们将每一条秒杀的请求存入消息队列（例如 RabbitMQ）中，放入消息队列后，给用户返回类似“抢购请求发送成功”的结果。而在消息队列中，我们将收到的下订单请求一个个的写入数据库中」，比起多线程同步修改数据库的操作，大大缓解了数据库的连接压力，最主要的好处就表现在数据库连接的减少：

同步方式：大量请求快速占满数据库框架开启的数据库连接池，同时修改数据库，导致数据库读写性能骤减。
异步方式：一条条消息以顺序的方式写入数据库，连接数几乎不变（当然，也取决于消息队列消费者的数量）。

「这种实现可以理解为是一中流量削峰：让数据库按照他的处理能力，从消息队列中拿取消息进行处理。」

结合之前的四篇秒杀系统文章，这样整个流程图我们就实现了：

代码实现

我们在源码仓库里，新增一个 controller 对外接口：

``` /* * 下单接口：异步处理订单 * @param sid * @return / @RequestMapping(value = "/createUserOrderWithMq", method = {RequestMethod.GET}) @ResponseBody public String createUserOrderWithMq(@RequestParam(value = "sid") Integer sid, @RequestParam(value = "userId") Integer userId) { try { // 检查缓存中该用户是否已经下单过 Boolean hasOrder = orderService.checkUserOrderInfoInCache(sid, userId); if (hasOrder != null && hasOrder) { LOGGER.info("该用户已经抢购过"); return "你已经抢购过了，不要太贪心....."; } // 没有下单过，检查缓存中商品是否还有库存 LOGGER.info("没有抢购过，检查缓存中商品是否还有库存"); Integer count = stockService.getStockCount(sid); if (count == 0) { return "秒杀请求失败，库存不足....."; }

    // 有库存，则将用户id和商品id封装为消息体传给消息队列处理
    // 注意这里的有库存和已经下单都是缓存中的结论，存在不可靠性，在消息队列中会查表再次验证
    LOGGER.info("有库存：[{}]", count);
    JSONObject jsonObject = new JSONObject();
    jsonObject.put("sid", sid);
    jsonObject.put("userId", userId);
    sendToOrderQueue(jsonObject.toJSONString());
    return "秒杀请求提交成功";
} catch (Exception e) {
    LOGGER.error("下单接口：异步处理订单异常：", e);
    return "秒杀请求失败，服务器正忙.....";
}

} ```

createUserOrderWithMq 接口整体流程如下：

检查缓存中该用户是否已经下单过：在消息队列下单成功后写入 Redis 一条用户 id 和商品 id 绑定的数据
没有下单过，检查缓存中商品是否还有库存
缓存中如果有库存，则将用户 id 和商品 id 封装为消息体 「传给消息队列处理」
注意：这里的 「有库存和已经下单」 都是缓存中的结论，存在不可靠性，在消息队列中会查表再次验证， 「作为兜底逻辑」

消息队列是如何接收消息的呢？我们新建一个消息队列，采用第四篇文中使用过的 RabbitMQ，我再稍微贴一下整个创建 RabbitMQ 的流程把：

pom.xml 新增 RabbitMq 的依赖：

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

写一个 RabbitMqConfig：

``` @Configuration public class RabbitMqConfig {

@Bean
public Queue orderQueue() {
    return new Queue("orderQueue");
}

} ```

添加一个消费者：

``` @Component @RabbitListener(queues = "orderQueue") public class OrderMqReceiver {

private static final Logger LOGGER = LoggerFactory.getLogger(OrderMqReceiver.class);

@Autowired
private StockService stockService;

@Autowired
private OrderService orderService;

@RabbitHandler
public void process(String message) {
    LOGGER.info("OrderMqReceiver收到消息开始用户下单流程: " + message);
    JSONObject jsonObject = JSONObject.parseObject(message);
    try {
        orderService.createOrderByMq(jsonObject.getInteger("sid"),jsonObject.getInteger("userId"));
    } catch (Exception e) {
        LOGGER.error("消息处理异常：", e);
    }
}

} ```

真正的下单的操作，在 service 中完成，我们在 orderService 中新建 createOrderByMq 方法：

``` @Override public void createOrderByMq(Integer sid, Integer userId) throws Exception {

Stock stock;
//校验库存（不要学我在trycatch中做逻辑处理，这样是不优雅的。这里这样处理是为了兼容之前的秒杀系统文章）
try {
    stock = checkStock(sid);
} catch (Exception e) {
    LOGGER.info("库存不足！");
    return;
}
//乐观锁更新库存
boolean updateStock = saleStockOptimistic(stock);
if (!updateStock) {
    LOGGER.warn("扣减库存失败，库存已经为0");
    return;
}

LOGGER.info("扣减库存成功，剩余库存：[{}]", stock.getCount() - stock.getSale() - 1);
stockService.delStockCountCache(sid);
LOGGER.info("删除库存缓存");

//创建订单
LOGGER.info("写入订单至数据库");
createOrderWithUserInfoInDB(stock, userId);
LOGGER.info("写入订单至缓存供查询");
createOrderWithUserInfoInCache(stock, userId);
LOGGER.info("下单完成");

} ```

真正的下单的操作流程为：

校验数据库库存
乐观锁更新库存（其他之前讲到的锁也可以啦）
写入订单至数据库
「写入订单和用户信息至缓存供查询」 ：写入后，在外层接口便可以通过判断 Redis 中是否存在用户和商品的抢购信息，来直接给用户返回“你已经抢购过”的消息。

「我是如何在 Redis 中记录商品和用户的关系的呢，我使用了 set 集合，key 是商品 id，而 value 则是用户 id 的集合，当然这样有一些不合理之处：」

这种结构默认了一个用户只能抢购一次这个商品
使用 set 集合，在用户过多后，每次检查需要遍历 set，用户过多有性能问题

大家知道需要做这种操作就好，具体如何在生产环境的 Redis 中存储这种关系，大家可以深入优化下。

@Override public Boolean checkUserOrderInfoInCache(Integer sid, Integer userId) throws Exception { String key = CacheKey.USER_HAS_ORDER.getKey() + "_" + sid; LOGGER.info("检查用户Id：[{}] 是否抢购过商品Id：[{}] 检查Key：[{}]", userId, sid, key); return stringRedisTemplate.opsForSet().isMember(key, userId.toString()); }

「整个上述实现只考虑最精简的流程，不把前几篇文章的限流，验证用户等加入进来，并且默认考虑的是每个用户抢购一个商品就不再允许抢购，我的想法是保证每篇文章的独立性和代码的任务最小化，至于最后的整合我相信小伙伴们自己可以做到。」

非异步与异步下单接口的性能对比

接下来就是喜闻乐见的 「非正规」 性能测试环节，我们来对异步处理和非异步处理做一个性能对比。

首先，为了测试方便，我把用户购买限制先取消掉，不然我用 Jmeter（JMeter 并发测试的使用方式参考秒杀系统第一篇文章）还要来模拟多个用户 id，太麻烦了，不是我们的重点。我们把上面的 controller 接口这一部分注释掉：

// 检查缓存中该用户是否已经下单过 Boolean hasOrder = orderService.checkUserOrderInfoInCache(sid, userId); if (hasOrder != null && hasOrder) { LOGGER.info("该用户已经抢购过"); return "你已经抢购过了，不要太贪心....."; }

这样我们可以用 JMeter 模拟抢购的情况了。

「我们先玩票大的！」 在我这个 1c4g1m 带宽的云数据库上， 「设置商品数量 5000 个，同时并发访问 10000 次」 。

服务器先跑起来，访问接口是 http://localhost:8080/createUserOrderWithMq?sid=1&userId=1

启动！

10000 个线程并发，直接把我的 1M 带宽小水管云数据库打穿了！

对不起对不起，打扰了，我们还是老实一点，不要对这么低配置的数据库有不切实际的幻想。

我们改成 1000 个线程并发，商品库存为 500 个， 「使用常规的非异步下单接口」 ：

对比 1000 个线程并发， 「使用异步订单接口」 ：

「可以看到，非异步的情况下，吞吐量是 37 个请求/秒，而异步情况下，我们的接只是做了两个事情，检查缓存中库存 + 发消息给消息队列，所以吞吐量为 600 个请求/秒。」

在发送完请求后，消息队列中立刻开始处理消息：

我截图了在 500 个库存刚刚好消耗完的时候的日志，可以看到，一旦库存没有了，消息队列就完成不了扣减库存的操作，就不会将订单写入数据库，也不会向缓存中记录用户已经购买了该商品的消息。

更加优雅的实现

那么问题来了，我们实现了上面的异步处理后，用户那边得到的结果是怎么样的呢？

用户点击了提交订单，收到了消息：您的订单已经提交成功。然后用户啥也没看见，也没有订单号，用户开始慌了，点到了自己的个人中心——已付款。发现居然没有订单！（因为可能还在队列中处理）

这样的话，用户可能马上就要开始投诉了！太不人性化了，我们不能只为了开发方便，舍弃了用户体验！

所以我们要改进一下，如何改进呢？其实很简单：

让前端在提交订单后，显示一个“排队中”， 「就像我们在小米官网抢小米手机那样」
同时，前端不断请求检查用户和商品是否已经有订单的接口，如果得到订单已经处理完成的消息，页面跳转抢购成功。

「是不是很小米（滑稽.jpg），暴露了我是 miboy 的事实」

实现起来，我们只要在后端加一个独立的接口：

/** * 检查缓存中用户是否已经生成订单 * @param sid * @return */ @RequestMapping(value = "/checkOrderByUserIdInCache", method = {RequestMethod.GET}) @ResponseBody public String checkOrderByUserIdInCache(@RequestParam(value = "sid") Integer sid, @RequestParam(value = "userId") Integer userId) { // 检查缓存中该用户是否已经下单过 try { Boolean hasOrder = orderService.checkUserOrderInfoInCache(sid, userId); if (hasOrder != null && hasOrder) { return "恭喜您，已经抢购成功！"; } } catch (Exception e) { LOGGER.error("检查订单异常：", e); } return "很抱歉，你的订单尚未生成，继续排队吧您嘞。"; }

我们来试验一下，首先我们请求两次下单的接口，大家用 postman 或者浏览器就好：

http://localhost:8080/createUserOrderWithMq?sid=1&userId=1

可以看到，第一次请求，下单成功了，第二次请求，则会返回已经抢购过。

因为这时候 Redis 已经写入了该用户下过订单的数据：

127.0.0.1:6379> smembers miaosha_v1_user_has_order_1 (empty list or set) 127.0.0.1:6379> smembers miaosha_v1_user_has_order_1 1) "1"

我们为了模拟消息队列处理茫茫多请求的行为，我们在下单的 service 方法中，让线程休息 10 秒：

``` @Override public void createOrderByMq(Integer sid, Integer userId) throws Exception {

// 模拟多个用户同时抢购，导致消息队列排队等候10秒
Thread.sleep(10000);

//完成下面的下单流程（省略）

} ```

然后我们清除订单信息，开始下单：

http://localhost:8080/createUserOrderWithMq?sid=1&userId=1

第一次请求，返回信息如上图。

紧接着前端显示排队中的时候，请求检查是否已经生成订单的接口，接口返回”继续排队“：

一直刷刷刷接口，10 秒之后，接口返回”恭喜您，抢购成功“，如下图：

整个流程就走完了。

结束语

这篇文章介绍了如何在保证用户体验的情况下完成订单异步处理的流程。内容其实不多，深度没有前一篇那么难理解。（我拖更也有一部分原因是因为我觉得上一篇的深度我很难随随便便达到，就不敢随意写文章，有压力。）

希望大家喜欢，目前来看，整个秒杀下订单的主流程我们全部介绍完了。当然里面很多东西都非常基础，比如数据库设计我一直停留在那几个破字段，比如订单的编号，其实不可能用主键 id 来做等等。

参考文献

基于微服务架构的秒杀系统服务端设计与实现（东南大学·徐冉）
基于Dubbo框架的开发者中心的设计与实现（北京交通大学·涂鹏程）
某订单秒杀系统的设计与实现（北京邮电大学·侯成程）
基于贪心算法的物流配送系统设计与实现（西北师范大学·柴荣）
基于微服务架构的秒杀系统服务端设计与实现（东南大学·徐冉）
基于微服务架构的秒杀系统服务端设计与实现（东南大学·徐冉）
基于Spring框架的短信通道系统的设计与实现（北京邮电大学·刘林华）
基于微服务架构之电商秒杀系统的设计与实现（华东师范大学·闫保正）
基于SSH框架的博客用户分享平台的设计与实现（河北工业大学·刘磊）
在线教育学习平台的设计与实现（华中科技大学·舒田浪）
基于微服务架构之电商秒杀系统的设计与实现（华东师范大学·闫保正）
基于Spring框架的短信通道系统的设计与实现（北京邮电大学·刘林华）
基于OAuth2.0协议的企业分布式授权系统设计与实现（华中科技大学·支猛）
基于微服务架构的秒杀系统服务端设计与实现（东南大学·徐冉）
基于SSH框架在线拍卖系统的分析与设计（北京邮电大学·陈旭）

本文内容包括但不限于文字、数据、图表及超链接等）均来源于该信息及资料的相关主题。发布者：毕设导航，原文地址：https://m.bishedaima.com/yuanma/35835.html

基于SpringBoot的简易秒杀系统

【秒杀系统】从零打造秒杀系统（一）：防止超卖

前言

秒杀系统

秒杀系统介绍

为什么要做所谓的“系统”

我们先从“防止超卖”开始吧

建立“简易”的数据库表结构

通过 HTTP 接口发起一次购买请求

Controller 层代码

Service 层代码

发起并发购买请求

避免超卖问题：更新商品库存的版本号

Controller 层

Service 层

Mapper

发起并发购买请求

源码

参考

【秒杀系统】零基础上手秒杀系统（二）：令牌桶限流 + 再谈超卖

前言

秒杀系统介绍

接口限流

接口限流实战：令牌桶限流算法

令牌桶算法与漏桶算法

使用 Guava 的 RateLimiter 实现令牌桶限流接口

Guava 中 RateLimiter 实现原理

再谈防止超卖

实现不需要版本号字段的乐观锁

实现悲观锁

悲观锁真的锁住库存了吗？

总结

参考

【秒杀系统】零基础上手秒杀系统（三）：抢购接口隐藏 + 单用户限制频率

前言

前文回顾和文章规划

项目源码在这里

抢购接口隐藏

代码逻辑实现

获取验证值接口

携带验证值下单接口

试验一下接口

单用户限制频率

使用 Redis/Memcached

试一试接口

能否不用 Redis/Memcached 实现用户访问频率统计

总结

参考

秒杀系统实战（四）| 缓存与数据库双写问题的争议

前言

前文回顾和文章规划

本篇文章主要内容

项目源码在这里

正文

缓存热点数据

为何要使用缓存

哪类数据适合缓存

缓存的利与弊

缓存和数据库双写一致性

不使用更新缓存而是删除缓存

先删除缓存，还是先操作数据库？

我一定要数据库和缓存数据一致怎么办

延时双删

删缓存失败了怎么办：重试机制

实战：先删除缓存，再更新数据库

实战：先更新数据库，再删缓存

实战：缓存延时双删

实战：删除缓存重试机制

实战：读取 binlog 异步删除缓存

扩展阅读

小结

结束语

参考

秒杀系统实战（五）| 如何优雅的实现订单异步处理

前言

本篇文章主要内容

前文回顾

项目源码

正文

秒杀系统介绍