SpringBoot实现并发、超发和锁机制/抢购示例：超发、乐观锁、悲观锁和Redis的使用

用户10175992

发布于 2022-11-15 13:23:13

8530

发布于 2022-11-15 13:23:13

文章被收录于专栏：辰远

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

1.1 数据库结构（MySQL）

DROP DATABASE IF EXISTS rush_to_purchase_db;
2
CREATE DATABASE rush_to_purchase_db;
3
USE rush_to_purchase_db;
4
5
/* 产品信息表 */
6
CREATE TABLE t_product(
7
    id INT(12) NOT NULL AUTO_INCREMENT COMMENT '商品编号',
8
    product_name VARCHAR(60) NOT NULL COMMENT '商品名称',
9
    stock INT(10) NOT NULL COMMENT '库存',
10
    price DECIMAL(16,2) NOT NULL COMMENT '单价',
11
    VERSION INT(10) NOT NULL DEFAULT 0 COMMENT '版本号',
12
    note VARCHAR(256) NULL COMMENT '备注',
13
    PRIMARY KEY(id)
14
);
15
/* 购买信息表 */
16
CREATE TABLE t_purchase_record(
17
  id INT(12) NOT NULL AUTO_INCREMENT COMMENT '记录编号',
18
  user_id INT(12) NOT NULL  COMMENT '用户编号',
19
  produce_id INT(12) NOT NULL  COMMENT '商品编号',
20
  price DECIMAL(16,2) NOT NULL COMMENT '价格',
21
  quantity INT(12) NOT NULL  COMMENT '数量',
22
  SUM DECIMAL(16,2) NOT NULL COMMENT '总价',
23
  purchase_time TIMESTAMP NOT NULL DEFAULT NOW() COMMENT '购买时间',
24
  note VARCHAR(512) NOT NULL COMMENT '备注',
25
  PRIMARY KEY(id)
26
);

1.2 创建SpringBoot的SSM项目，实现购买Action功能

（1）Mapper

public interface ProductMapper {
2
    @Select("SELECT id,product_name AS productName,stock,price,VERSION,note FROM t_product where id=#{id}")
3
    Product findById(long id);
4
    @Update("update t_product set stock=stock- #{quantity} where id=#{id}")
5
    void descreaseStock(@Param("id")long id, @Param("quantity")long quantity);
6
}
7
8
public interface PurchaseRecordMapper {
9
    @Options(keyProperty = "id", useGeneratedKeys = true)
10
    @Insert("INSERT INTO t_purchase_record(user_id,produce_id,price,quantity,SUM,purchase_time,note) VALUES(#{userId},#{productId},#{price},#{quantity},#{sum},#{purchaseTime},#{note})")
11
    void add(PurchaseRecord record);
12
}

（2）Service

@Service
2
public class PurchaseServiceImpl implements PurchaseService {
3
    @Autowired
4
    private ProductMapper productDb;
5
    @Autowired
6
    private PurchaseRecordMapper recordDb;
7
    @Transactional
8
    public boolean purchase(long userId, long productId, long quantity) {
9
        Product product = productDb.findById(productId);    //查库存
10
        if(product.getStock()<quantity) {       //如果商品库存少于购买数量
11
            return false;                       //返回失败
12
        }
13
        productDb.descreaseStock(productId, quantity);      //减库存
14
        PurchaseRecord record = initPurchaseRecord(userId, product, quantity);
15
        recordDb.add(record);                               //保存购买记录
16
        return true;
17
    }
18
    private PurchaseRecord initPurchaseRecord(long userId, Product product, long quantity) {
19
        PurchaseRecord pr = PurchaseRecord.builder()
20
                .userId(userId)
21
                .productId(product.getId())
22
                .price(product.getPrice())
23
                .quantity(quantity)
24
                .sum(product.getPrice()*quantity)
25
                .purchaseTime(new Timestamp(System.currentTimeMillis()))
26
                .note("")
27
                .build();
28
        return pr;
29
    }
30
}

（3）Controller

@Controller
2
public class PurchaseController {
3
    @Autowired
4
    private ProductService productService;
5
    @Autowired
6
    private PurchaseService purchaseService;
7
    @GetMapping("/index")           //jsp测试页
8
    public String index() {         
9
        return "index";
10
    }
11
    @PostMapping("/api/purchase")   //处理抢购请求
12
    @ResponseBody
13
    public JsonResult purchase(long userId, long productId, long quantity) {
14
        boolean ok = purchaseService.purchase(userId, productId, quantity);
15
        return new JsonResult(ok, ok?"抢购成功":" 抢购失败");
16
    }
17
}
18
19
@Data
20
@AllArgsConstructor
21
@NoArgsConstructor
22
class JsonResult{
23
    private boolean ok;
24
    private String message;
25
}

（4）使用jQuery Ajax模拟抢购过程

<script src="static/jquery-3.4.1.min.js"></script>
2
<script>
3
    $(function(){
4
        $("#rush2buy").click(function(){    //#rush2buy 是模拟抢购的按钮
5
            for(var i=1; i<=400; i++){      //循环执行500次
6
                var params = {userId:1, productId:1, quantity: 3};
7
                $.post("api/purchase", params
8
                        , function(result){
9
                            console.log(new Date().getTime());  //记录每次执行完的时间
10
                        }
11
                );
12
            }
13
        });
14
    })
15
</script>

数据库发生超发现象：

2 线程同步方案

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

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

@Service
2
public class PurchaseServiceImpl implements PurchaseService {
3
    ......
4
    @Transactional
5
    public synchronized boolean purchase(long userId, long productId, long quantity) {
6
        Product product = productDb.findById(productId);
7
        if(product.getStock()<quantity) {
8
            return false;
9
        }
10
        productDb.descreaseStock(productId, quantity);
11
        PurchaseRecord record = initPurchaseRecord(userId, product, quantity);
12
        recordDb.add(record);
13
        return true;
14
    }
15
}

线程同步把抢购业务方法变成了单线程执行，能保证不会发生超发现象，但随着并发量增加性能下降较大。

3 数据库“悲观锁”方案

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

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

public interface ProductMapper {
2
    @Select("SELECT id,product_name AS productName,stock,price,VERSION,note FROM t_product where id=#{id} for update")
3
    Product findById(long id); 
4
    ......
5
}

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

经过上述修改，并发执行后就不会超发了。

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

4 “乐观锁”方法

（1）乐观锁的概念

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

（2）乐观锁的使用

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

public interface ProductMapper {
2
    //不使用悲观锁
3
    @Select("SELECT id,product_name AS productName,stock,price,VERSION,note FROM t_product where id=#{id}")
4
    Product findById(long id);  
5
    //2 乐观锁
6
    @Update("update t_product set stock=stock- #{quantity} version=version+1 where id=#{id} and version=#{version}")
7
    int descreaseStock(@Param("id")long id, @Param("quantity")long quantity, @Param("version")long version);
8
}

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

@Service
2
public class PurchaseServiceImpl implements PurchaseService {
3
    ......
4
5
    //2 乐观锁
6
    @Transactional
7
    public boolean purchase(long userId, long productId, long quantity) {
8
        Product product = productDb.findById(productId);
9
        if(product.getStock()<quantity) {
10
            return false;
11
        }
12
        long version = product.getVersion();    //获取旧数据的版本号
13
        int result = productDb.descreaseStock(productId, quantity, version);    //把版本号投入减库存方法
14
        if(result==0) {                         // 查看按照该版本号能否修改库存，不能表示版本号已经变更，有其他并发请求修改了库存，放弃当前操作
15
            return false;
16
        }
17
        PurchaseRecord record = initPurchaseRecord(userId, product, quantity);
18
        recordDb.add(record);
19
        return true;
20
    }
21
}

乐观锁可以很好的提高执行效率，也可以确保不会出现超发的数据不一致问题，

但是，乐观锁也有自己的问题，请求失败率变得非常高，以致数据库还有剩余的商品！

实际中，我们需要在出现版本不一致的时候，终止当前事务同时再次引发一个新的购买事务，在一定次数（时间）范围反复尝试。以下演示的是限定次数的乐观锁：

@Service
2
public class PurchaseServiceImpl implements PurchaseService {
3
    ......
4
    @Transactional
5
    public boolean purchase(long userId, long productId, long quantity) {
6
        for(int i=1; i<=3; i++){    //限定次数的乐观锁
7
            Product product = productDb.findById(productId);
8
            if(product.getStock()<quantity) {
9
                return false;
10
            }
11
            long version = product.getVersion();
12
            int result = productDb.descreaseStock(productId, quantity, version);
13
            if(result==0) {
14
                continue;           //单次失败，再次执行
15
            }
16
            PurchaseRecord record = initPurchaseRecord(userId, product, quantity);
17
            recordDb.add(record);
18
            return true;
19
        }
20
        return false;
21
    }

这样会增加成功的几率：

5 使用Redis解决高并发超发

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

Redis中有很多可以解决并发问题的技术：例如原子计数器、分布式锁、原子性的Lua脚本等等。这里介绍一个简单的方案“原子计数器”来减少。

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 依赖

<!-- redis -->
2
        <dependency>
3
            <groupId>org.springframework.boot</groupId>
4
            <artifactId>spring-boot-starter-data-redis</artifactId>
5
        </dependency>

（2）修改 application.yml 配置Redis

spring:
2
  #redis配置连接
3
  redis:
4
    database: 0
5
    host: localhost
6
    port: 6379
7
    password: 1234
8
    timeout: 3600000 #缓存一个小时

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

这里为了方便测试，直接用SpringBootTest来模拟：

@RunWith(SpringRunner.class)
2
@SpringBootTest
3
public class AddStocks2RedisTests {
4
    @Autowired
5
    private RedisTemplate<String, String> redisTemplate;
6
    @Autowired
7
    private ProductService productService;
8
    
9
    @Test
10
    public void testAddStocks2Redis() {
11
        productService.findAll().forEach(x->{
12
            redisTemplate.opsForHash().put("product-stocks", x.getId()+"", x.getStock()+"");
13
        });
14
        redisTemplate.expire("product-stocks", 3600, TimeUnit.SECONDS);
15
    }
16
}

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

// 4、Redis 原子计数器方案
2
    @Autowired
3
    private RedisTemplate<String, String> redisTemplate;
4
    @Transactional
5
    public boolean purchase(long userId, long productId, long quantity) {
6
        //从redis获取某商品库存
7
        long stock = Long.parseLong(redisTemplate.opsForHash().get("product-stocks", productId+"").toString()); 
8
        //从redis查询该商品的销量 sales，如果还能购买，则在redis的原子计数器
9
        synchronized (PurchaseService.class) {
10
            //从redis中读取某商品的销量，比如key为“product-sales-123”
11
            String value = redisTemplate.opsForValue().get("product-sales-"+productId);
12
            long sales = 0;
13
            if(value!=null) {
14
                sales = Long.valueOf(value);
15
            }else {     
16
                //如果redis中没有该商品的销量，则初始化为0，设定一定的过期时间
17
                redisTemplate.opsForValue().set("product-sales-"+productId, "0");
18
                redisTemplate.expire("product-sales-"+productId, 3600, TimeUnit.SECONDS);
19
            }
20
            //判断 如果 库存 < 销量+本次购买数量 则不能销售，返回 false
21
            if( stock < (sales+quantity) ) {
22
                return false;
23
            }
24
            //增加库存量，原子计数器
25
            redisTemplate.opsForValue().increment("product-sales-"+productId, quantity);
26
        }
27
        //完成余下的数据库操作，保存数据，减少库存和增加销售记录
28
        Product product = productDb.findById(productId);    //查商品
29
        productDb.descreaseStock(productId, quantity);      //减库存
30
        PurchaseRecord record = initPurchaseRecord(userId, product, quantity);
31
        recordDb.add(record);                               //添加销售记录
32
        return true;
33
    }

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

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

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2021-12-24，如有侵权请联系 cloudcommunity@tencent.com 删除

云数据库 Redis