Redis解决高并发问题

1 模拟商品抢购和并发的效果

这里模拟一个商品抢购的过程所带来的问题，以及解决问题的思路。

这里模拟的商品抢购过程是一个商品正常购买的过程，其中包含了两个主要的步骤：商品库存减少和商品购买记录的添加。

下面搭建项目环境。

1.1 数据库结构（MySQL）

DROP DATABASE IF EXISTS rush_to_purchase_db;CREATE DATABASE rush_to_purchase_db;USE rush_to_purchase_db;/* 产品信息表 */CREATE TABLE t_product(id INT(12) NOT NULL AUTO_INCREMENT COMMENT '商品编号',NAME VARCHAR(60) NOT NULL COMMENT '商品名称',stock INT(10) NOT NULL COMMENT '库存',price DECIMAL(16,2) NOT NULL COMMENT '单价',VERSION INT(10) NOT NULL DEFAULT 0 COMMENT '版本号',note VARCHAR(256) NULL COMMENT '备注',PRIMARY KEY(id));/* 购买信息表 */CREATE TABLE t_purchase_record(id INT(12) NOT NULL AUTO_INCREMENT COMMENT '记录编号',userId INT(12) NOT NULLCOMMENT '用户编号',productId INT(12) NOT NULLCOMMENT '商品编号',price DECIMAL(16,2) NOT NULL COMMENT '价格',quantity INT(12) NOT NULLCOMMENT '数量',purchaseTime TIMESTAMP NOT NULL DEFAULT NOW() COMMENT '购买时间',note VARCHAR(512) NOT NULL COMMENT '备注',PRIMARY KEY(id));INSERT INTO t_product VALUES(1, 'Yogas2020笔记本电脑', 50, 4000, DEFAULT, 'Yogas2020笔记本电脑，14.3寸，轻便之选');

1.2 创建SpringBoot的SSM项目，实现基本购物功能

（1）Model

public class Product implements Serializable {private int id;private String name;private int stock;private double price;private int version;private String note;//省略getter、setter}

public class PurchaseRecord implements Serializable {private int id;private int userId;private int productId;private double price;private int quantity;private double totalPrice;private Timestamp purchaseTime;private String note; //省略getter、setter }

（2）Mapper

public interface ProductMapper {@Select("SELECT id,name,stock,price,VERSION,note FROM t_product where id=#{id}")Product selectById(long id);@Update("update t_product set stock=stock- #{quantity} where id=#{id}")void descreaseStock(long id, long quantity);}

public interface PurchaseRecordMapper {@Options(keyProperty = "id", useGeneratedKeys = true)@Insert("INSERT INTO t_purchase_record(userId,productId,price,quantity,purchaseTime,note) VALUES(#{userId},#{productId},#{price},#{quantity},#{purchaseTime},#{note})")void insert(PurchaseRecord record);}

（3）Service

@Servicepublic class PurchaseServiceImpl implements PurchaseService {@Autowiredprivate PurchaseRecordMapper purchaseRecordMapper;@Autowiredprivate ProductMapper productMapper;@Transactionalpublic boolean purchase(int userId, int productId, int quantity) {//根据产品id判断库存是否够Product product= productMapper.selectById(productId);//如果库存不够，购买失败if(quantity>product.getStock()) {return false;}//如果库存足够--减库存productMapper.descreaseStock(productId,quantity);//增加购买记录addPurchaseRecord(userId,product,quantity);return true;}//添加购买记录private void addPurchaseRecord(int userId, Product product, int quantity){PurchaseRecord record=new PurchaseRecord();record.setPrice(product.getPrice());record.setPurchaseTime(new Timestamp(System.currentTimeMillis()));record.setProductId(product.getId());record.setUserId(userId);record.setNote("购买时间:"+System.currentTimeMillis());purchaseRecordMapper.insert(record);}}

（4）Controller

@RestControllerpublic class PurchaseController {@Autowiredprivate PurchaseService purchaseService;@PostMapping("/api/purchase")public String purchase(int userId,int productId,int quantity){boolean flag=purchaseService.purchase(userId,productId,quantity);return flag?"抢购成功":"抢购失败";}}

（4）index.html：使用jQuery Ajax模拟抢购过程

<script src="jquery.js"></script><script>$(function(){//抢购按钮模拟500人抢购50台笔记本$("#rush2buy").click(function(){for(var i=1; i<=500; i++){var params = {userId:1, productId:1, quantity: 1};$.post("api/purchase", params, function(result){console.log(new Date().getTime());});}});})</script>

数据库发生超发现象：

注意：

如果是低并发量测试一般时没问题的，如果购买不成功有正确的提示，如果是高并发量就会出现超发现象，即库存小于0的问题。即库存原本只有50，但是500个人去抢的时候，最后库存变成了-3,相当于卖出了53台.

2 方案1：线程同步

上述的超发现象，归根到底在于数据库时被多个线程同时访问的，在没有加锁的情况下，上述代码并不是线程安全的。

最简单的办法是为业务方法添加线程同步“synchroized”关键字，确保同一个时间只有一个线程进入操作。

//添加synchronized实现线程同步@Transactionalpublic synchronized boolean purchase(int userId, int productId, int quantity) {//根据产品id判断库存是否够Product product= productMapper.selectById(productId);//如果库存不够，购买失败if(quantity>product.getStock()) {return false;}//如果库存足够--减库存productMapper.descreaseStock(productId,quantity);//增加购买记录addPurchaseRecord(userId,product,quantity);return true;}

线程同步把抢购业务方法变成了单线程执行，能保证不会发生超发现象，但随着并发量增加性能下降较大。
如果还是发生超发现象 ，是电脑性能比较强 ，我的代码逻辑上还存在缺陷，高并发造成瞬间数据穿透。

3 方案2：数据库“悲观锁”

高并发情况下出现的问题，主要原因在于共享资源（stock）被多个线程并行修改。

如果一个数据库事务读取到产品库存后，就直接把该行数据锁定，不允许其它事务读写，直到事务完成商品库存的减少在释放锁，就不会出现并超发现象了。MySQL就提供数据库锁的解决方案，这种锁称为悲观锁。具体操作如下：

修改上述Mapper中的查询语句，在每次查询商品库存的时候加上更新锁。

public interface ProductMapper {@Select("SELECT id,name AS productName,stock,price,VERSION,note FROM t_product where id=#{id} for update")Product selectById(long id); ......}

注意上述语句中“SELECT id,product_name AS productName,stock,price,VERSION,note FROM t_product where id=#{id} for update” 中的**“for update”称为更新锁**，在数据库事务执行过程中，它会锁定查询出来的数据，其他事务不能再对其进行读写，直到该事务完成才会只放锁，这样能避免数据不一致了。

经过上述修改，并发执行后就不会超发了。
如果还是发生超发现象 ，是电脑性能比较强 ，我的代码逻辑上还存在缺陷，高并发造成瞬间数据穿透。

但由于加锁，会导致实际代码的执行时间有所增加。

4 方案3：“乐观锁”

（1）乐观锁的概念

悲观锁虽然可以解决高并发下的超发现象，却并非高效方案，另一些开发者会采用乐观锁方案。乐观锁并非数据库加锁和阻塞的解决方案，乐观锁把读取到的旧数据保存下来，等到要对数据进行修改的时候，会先把旧数据与当前数据库数据进行比较，如果旧数据与当前数据一致，我们就认为数据库没有被并发修改过，否则就认为数据已经被其它并发请求修改，当前的事务回滚，不再修改任何数据。在实际操作中，乐观锁通常需要在数据表中增加“数据版本号”这样一个字段，以标识当前数据和旧数据是否一致，每次修改数据后“数据版本号”要增加。

（2）乐观锁的使用

修改减少库存的Mapper方法，每次减少库存的时候同时修改数据的版本号version

public interface ProductMapper {//不使用悲观锁@Select("SELECT id,name AS productName,stock,price,VERSION,note FROM t_product where id=#{id}")Product selectById(long id);//使用乐观锁(添加版本号条件和版本号增加)@Update("update t_product set stock=stock- #{quantity}, version=version+1 where id=#{id} and version=#{version}")void descreaseStock(long id, long quantity, long version);}

修改业务方法，每次修改库存时检查是否修改到，如果没改到数据“result==0”则表示数据版本号已经变更，有其他并发请求改过库存，放弃当前操作。

@Servicepublic class PurchaseServiceImpl implements PurchaseService {......//乐观锁方案@Transactionalpublic boolean purchase(int userId, int productId, int quantity) {//根据产品id判断库存是否够Product product=productMapper.selectById(productId);//如果库存不够，购买失败if(quantity>product.getStock()) {return false;}//如果库存足够--减库存int result=productMapper.descreaseStock(productId,quantity,product.getVersion());// 影响行数0，没修改成，代表版本号已经改变，已经并发，放弃本次修改System.out.println(result);if(result==0) {return false;}//增加购买记录addPurchaseRecord(userId,product,quantity);return true;}}

乐观锁可以很好的提高执行效率，也可以确保不会出现超发的数据不一致问题。但是，乐观锁也有自己的问题，请求失败率变得很高，以致数据库可能还有剩余的商品。

例如，我们把模拟的抢购人数从500将为100，则可能看到库存还有剩余商品。

for(var i=1; i<=100; i++){//将为100var params = {userId:1, productId:1, quantity: 1};$.post("api/purchase", params, function(result){console.log(new Date().getTime());});}

因此，乐观锁虽然能避免并发，却并不适合抢购的业务场景，当然，我们也可以增加失败重试的机制去增加成功率。

5 方案4：使用Redis提高并发性

实际中，引入Redis这类NoSQL是提高并发性的更好选择。

Redis这类的NoSQL数据库以Hash的方式把数据存放在内存中，在理想环境中每秒读写次数可以高达10万次，数据吞吐效率远高于SQL数据库，因此常常用来解决大规模并发的访问效率问题。

Redis 的“INCR”命令可以将key中存储的数字值加1。如果key不存在，那么Key的值会先被初始化为0，然后再执行INCR操作。Redis 中的该操作是原子性的，不会被高并发打断，可以确保数据的一致性。

5.1 使用Redis计数器的处理思路：

（1）抢购开始前，Redis缓存抢购商品的HashMap：从数据库中读取参加抢购的商品（ID）和对应的库存（stock）保存在Redis中；

（2）Redis中为每件抢购商品单独保存一个计数器：key中保存商品id信息，value中保存商品的销量（sales）；

（3）处理每个购买请求时，先从Redis中读取商品库存（stock）和之前的销量（sales）；若“库存<之前销量+本次购买量” 则 返回购买失败；否则 使用原子计数器增加销量，并继续执行后续的数据库操作；

5.2 具体实现：

（1）为Spring Boot 项目引入 Redis 依赖

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

（2）修改 application.yml 配置Redis

spring:#redis配置连接redis:database: 0host: localhostport: 6379

（4）在开始抢购前缓存商品和库存集合

这里为了方便测试，直接用SpringBootTest来模拟把商品预先缓存到Redis的操作。

@RunWith(SpringRunner.class)@SpringBootTestpublic class AddStocks2RedisTests {@Autowired@Qualifier("stringRedisTemplate")private RedisTemplate redisTemplate;@Autowiredprivate ProductService productService;@Testpublic void testAddStocks2Redis() {productService.findAll().forEach(x->{redisTemplate.opsForHash().put("product-stocks", x.getId()+"", x.getStock()+"");});redisTemplate.expire("product-stocks", 3600, TimeUnit.SECONDS);}}

（4）重写PurchaseServiceImpl中的purchase方法，处理购买前先检查Redis中的库存和销量

@Autowired@Qualifier("stringRedisTemplate")private RedisTemplate<String,String> redisTemplate;//使用Redis判断库存量是否够发，过滤掉超发请求，然后再进行SQL操作@Transactionalpublic boolean purchase(int userId, int productId, int quantity) {/* 使用Redis对比并发时的销量和库存量是否一致，排除超发请求 *///读取商品库存long stock = Long.parseLong(redisTemplate.opsForHash().get("product-stocks", productId + "").toString());//读取商品销量String value = redisTemplate.opsForValue().get("product-sales-" + productId);int sales = 0;if (value != null) {sales = Integer.valueOf(value);} else {//若还没有对应产品的销量，向Redis初始化该产品销量为0redisTemplate.opsForValue().set("product-sales-" + productId, "0", 3600, TimeUnit.SECONDS);}if (stock < (sales + quantity)) { //对比库存量和销量，库存不足销售时返回falsereturn false;}redisTemplate.opsForValue().increment("product-sales-" + productId, quantity);//增加Redis中的销量/* 以下为MySQL数据库操作，不超发的请求才执行，可以保留悲观锁操作，以防Redis中还可能有漏网的并发*/Product product = productMapper.selectById(productId);if (product.getStock() < quantity) {return false;}//减少库存productMapper.descreaseStock(productId, quantity);//增加购买记录addPurchaseRecord(userId,product,quantity);return true;}

这个方法利用了Redis的高速访问特性，有效的提高了并发超发的检查效率。

下为MySQL数据库操作，不超发的请求才执行，可以保留悲观锁操作，以防Redis中还可能有漏网的并发 */
Product product = productMapper.selectById(productId);
if (product.getStock() < quantity) {
return false;
}
//减少库存
productMapper.descreaseStock(productId, quantity);
//增加购买记录
addPurchaseRecord(userId,product,quantity);
return true;
}

这个方法利用了Redis的高速访问特性，有效的提高了并发超发的检查效率。在实际应用中，我们还可以把购买的整个过程使用Redis操作记录下来，在空闲的时候再把结果同步回SQL数据库，这样就真的能解决并发的效率问题了。