在高并发场景下进行减库存，该应用程序是分布式部署，使用nginx做负载均衡，使用redis做分布式锁
使用Spirngboot开发测试
- 1.首先搭建环境
pom文件
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.redis</groupId>
<artifactId>test-redis-lock</artifactId>
<version>1.0-SNAPSHOT</version>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.2.RELEASE</version>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.5.0</version>
</dependency>
</dependencies>
</project>
项目目录
项目目录
application.yaml文件
server:
port: 8080 #服务端口
spring:
redis:
host: 180.76.244.226 #redis服务器地址
启动器
@SpringBootApplication
public class TestRedisApplication {
public static void main(String[] args) {
SpringApplication.run(TestRedisApplication.class);
}
- 2.编写减库存代码
package com.test.controller;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* @author lipeng
* 时间： 2020-10-13 10:00
* 描述：
*/
@RestController
@RequestMapping("/api/demo")
public class Demo2Controller {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@RequestMapping("/deduct_lock")
public String deductLock() {
synchronized (this) {
//加锁
Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));//根据key 获取redis中的值
if (stock > 0) {
int realStock = stock - 1;
//将值再存入redis中
stringRedisTemplate.opsForValue().set("stock", realStock + "");
System.out.println("减库存成功,剩余库存数:" + realStock);
} else {
System.out.println("当前库存数不足");
}
}
return "end";
}
}
-
3.分布式部署项目
由于使用的是Springboot，所以分布式部署还是很方便的，我们只需改变端口号，并重新运行一遍main函数即可
分布式部署
nginx负载均衡配置
upstream redislock{
server 192.168.110.99:8080 weight=1;
server 192.168.110.99:8081 weight=1;
}
server{
listen 9000;
server_name localhsot;
location / {
root html;
index index.html index.html;
proxy_pass http://redislock;
}
}
- 4.使用压力测试进行jmter进行测试
jmter
我这里配置了200的并发量
http请求
之后就会发起请求,我们发现出现很多超卖请求
。。。。。。。。要给redis中存一个值，stock 我存的50，这里就不截图了 - 5.使用reids实现分布式锁，并总结遇到的坑
package com.test.controller;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* @author lipeng
* 时间： 2020-10-12 8:59
* 描述：高并发下减库存的实现
*/
@RestController
@RequestMapping("/api/demo")
public class DemoController {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Autowired
private Redisson redisson;
@RequestMapping("/deduct_lock")
public String deductLock() {
String lockKey = "lockKey";
/* Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "lipeng"); //使用redis实现1.0分布式锁
stringRedisTemplate.expire(lockKey,10, TimeUnit.SECONDS); //10s超时*/
/* String clientId= UUID.randomUUID().toString(); //创建标识，判断是否是自己还持有锁
//底层为原子块执行
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 10, TimeUnit.SECONDS);
if (!result) {
return "error";
}*/
/**
* -------------------------------------redis实现1.0版本的分布式锁 bug 总结--------------------------------*
* 1.当第一个线程进入时，拿到了锁，但是在执行过程中抛出了一个异常，这个时候锁并没有释放 产生死锁现象.
* ----解决方案：加上一个 try{} finally{} 使程序不管是否抛出异常 锁必须释放
*2.在应用执行过程中，服务宕机了，此时该应用已经拿到锁，但是并没有释放，之后别的机器上过来的请求依然会直接返回，再次产生死锁状态
*----解决方案：给这个锁的key加一个超时时间，10s后如果该key存在，直接自动销毁 stringRedisTemplate.expire(lockKey,10, TimeUnit.SECONDS);
* 3.在程序执行到给key加超时时间时 程序宕机了，又会产生死锁
*----解决方案：StringRedisTemplate提供了一个api：stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "lipeng", 10, TimeUnit.SECONDS)；
* 4.在高并发场景下，由于程序执行时间超出失效时间导致锁失效问题
*----解决方案：创建一个标识，在释放锁的时候，判断是不是自己设置的锁，如果是就释放
* 5.在高并发场景下，锁的超时时间解决
*----解决方案：使用redisson来实现分布式锁（在加锁成功后，在后台开启一个线程，实现定时任务，每隔10秒检查是否还持有锁如果持有则延长锁的时间）
*/
//获取锁
RLock redissonLock = redisson.getLock(lockKey);
try {
//加锁
redissonLock.lock();
Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));//根据key 获取redis中的值
if (stock > 0) {
int realStock = stock - 1;
//将值再存入redis中
stringRedisTemplate.opsForValue().set("stock", realStock + "");
System.out.println("减库存成功,剩余库存数:" + realStock);
} else {
System.out.println("当前库存数不足");
}
} finally {
//释放锁
redissonLock.unlock();
/*if(clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))){ //判断是否是自己加的锁，如果是自己加的锁就释放
stringRedisTemplate.delete(lockKey);
}*/
}
return "end";
}
}
而最终我们选择使用redisson,该框架对redis进行了再次封装，它的使用场景也多为分布式。
redisson分布式锁实现原理
