---

typora-copy-images-to: images

1. 库存超卖问题

现在创建一个库存里面有100个商品，然后对这个商品进行购买。springboot 整合 redis实现

pom.xml

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.2.6.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.loser</groupId><artifactId>redis-lock</artifactId><version>0.0.1-SNAPSHOT</version><name>redis-lock</name><description>redis-lock</description><properties><java.version>1.8</java.version></properties><dependencies><!--web+actuator 图形化支持--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><!--集成redis--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-redis</artifactId><version>1.4.1.RELEASE</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.3</version></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId></dependency><!--redisson--><dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.13.4</version></dependency><!--通用配置--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

RedisConfig.java

    @Beanpublic RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {RedisTemplate<String, Object> template = new RedisTemplate<>();RedisSerializer<String> redisSerializer = new StringRedisSerializer();Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);ObjectMapper om = new ObjectMapper();om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);jackson2JsonRedisSerializer.setObjectMapper(om);template.setConnectionFactory(factory);
//key序列化方式template.setKeySerializer(redisSerializer);
//value序列化template.setValueSerializer(jackson2JsonRedisSerializer);
//value hashmap序列化template.setHashValueSerializer(jackson2JsonRedisSerializer);return template;}@Beanpublic CacheManager cacheManager(RedisConnectionFactory factory) {RedisSerializer<String> redisSerializer = new StringRedisSerializer();Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
//解决查询缓存转换异常的问题ObjectMapper om = new ObjectMapper();om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);jackson2JsonRedisSerializer.setObjectMapper(om);
// 配置序列化（解决乱码的问题）,过期时间600秒RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig().entryTtl(Duration.ofSeconds(600)).serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer)).serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer)).disableCachingNullValues();RedisCacheManager cacheManager = RedisCacheManager.builder(factory).cacheDefaults(config).build();return cacheManager;}

GoodController.java

            String goods = redisTemplate.opsForValue().get(proName);int goodsNumber = goods == null ? 0 : Integer.parseInt(goods);if (goodsNumber > 0) {//减去之后的库存int resultNumber = goodsNumber - 1;redisTemplate.opsForValue().set(proName, String.valueOf(resultNumber));System.out.println("成功买到商品，库存还剩下：" + resultNumber + "\t服务提供端口：" + serverPort);return "成功买到商品，库存还剩下：" + resultNumber + "\t服务提供端口：" + serverPort;} else {System.out.println("商品已售完/活动结束/调用超时，欢迎下次光临" + "\t服务提供端口：" + serverPort);}return "商品已售完/活动结束/调用超时，欢迎下次光临" + "\t服务提供端口：" + serverPort;

准备两个服务

搭建的服务架构

进行测试

超卖的现象是必然的

2. 发现单机的服务，会存在许多问题，比如说数据的原子性不会保证 ，拿到的数据可能会有重复

给单机线程加锁进行测试

        synchronized (this){String goods = redisTemplate.opsForValue().get(proName);int goodsNumber = goods == null ? 0 : Integer.parseInt(goods);if (goodsNumber > 0) {//减去之后的库存int resultNumber = goodsNumber - 1;redisTemplate.opsForValue().set(proName, String.valueOf(resultNumber));System.out.println("成功买到商品，库存还剩下：" + resultNumber + "\t服务提供端口：" + serverPort);return "成功买到商品，库存还剩下：" + resultNumber + "\t服务提供端口：" + serverPort;} else {System.out.println("商品已售完/活动结束/调用超时，欢迎下次光临" + "\t服务提供端口：" + serverPort);}return "商品已售完/活动结束/调用超时，欢迎下次光临" + "\t服务提供端口：" + serverPort;}

两个服务代码是一样的

jmeter压力测试

准备1000个线程对100个库存进行测试 一次执行

两个线程同时消费 造成了买一送一的现象哈哈。

3. 再次升级— 添加分布式锁 redis 自带分布式锁

setnx 命令给当前程序添加分布式锁

 private static final String proName = "goods:1001";private final Lock lock = new ReentrantLock();private final String redisLock = "lock";@GetMapping("/by_goods")public String byGoods() {String value = UUID.randomUUID().toString()+ Thread.currentThread().getName();//====setnx  加锁Boolean flag = redisTemplate.opsForValue().setIfAbsent(redisLock, value);if (flag==false){return "拿锁失败";}String goods = redisTemplate.opsForValue().get(proName);int goodsNumber = goods == null ? 0 : Integer.parseInt(goods);if (goodsNumber > 0) {//减去之后的库存int resultNumber = goodsNumber - 1;redisTemplate.opsForValue().set(proName, String.valueOf(resultNumber));//删除key 解锁redisTemplate.delete(redisLock);System.out.println("成功买到商品，库存还剩下：" + resultNumber + "\t服务提供端口：" + serverPort);return "成功买到商品，库存还剩下：" + resultNumber + "\t服务提供端口：" + serverPort;} else {System.out.println("商品已售完/活动结束/调用超时，欢迎下次光临" + "\t服务提供端口：" + serverPort);}return "商品已售完/活动结束/调用超时，欢迎下次光临" + "\t服务提供端口：" + serverPort;}

但是还有个问题这里释放锁，如果程序执行期间异常了，可能会造成死锁，所以可以把释放锁卸载finally块里面

可以想修改成这样，但是这样还不是最完美的，如果执行期间机房着火了（哈哈），等于说这个程序执行一般挂了，redis还是没有得到释放，基于这个问题可以给锁设置过期时间

尽管过程机房失火了或者电脑爆炸了，程序在执行期间都会自动释放锁，到了10秒后就过期了，这里切记不能分开设置，分开设置依旧不是原子性的操作，有可能设置完锁之后程序就不能执行了，以为这样完美了么，如果当前程序的执行时间大于10s呢，虽然锁过期了，但是你可能在删除key的时候将其他的线程的key给删除了，所以这样依旧不是完美的，基于业务我们可以做一下修改

但是这样依旧不行 ，相信大家看出来了，删除也不是原子性的，基于这个redis推荐我们使用lua脚本来实现判断删除原子性的问题,除了lua脚本还有其他的一种方案，使用redis 本身自带的事务，我们来看一下redis命令

4. redis事务

MULTI 、 EXEC 、 DISCARD 和 WATCH 是 Redis 事务相关的命令。事务可以一次执行多个命令， 并且带有以下两个重要的保证：

• 事务是一个单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。
• 事务是一个原子操作：事务中的命令要么全部被执行，要么全部都不执行。

EXEC 命令负责触发并执行事务中的所有命令：

• 如果客户端在使用 MULTI 开启了一个事务之后，却因为断线而没有成功执行 EXEC ，那么事务中的所有命令都不会被执行。
• 另一方面，如果客户端成功在开启事务之后执行 EXEC ，那么事务中的所有命令都会被执行。

当使用 AOF 方式做持久化的时候， Redis 会使用单个 write(2) 命令将事务写入到磁盘中。

然而，如果 Redis 服务器因为某些原因被管理员杀死，或者遇上某种硬件故障，那么可能只有部分事务命令会被成功写入到磁盘中。

如果 Redis 在重新启动时发现 AOF 文件出了这样的问题，那么它会退出，并汇报一个错误。

使用redis-check-aof程序可以修复这一问题：它会移除 AOF 文件中不完整事务的信息，确保服务器可以顺利启动。

简单的思路来说：

java修改的代码如下

@RestController
public class GoodController {@Autowiredprivate StringRedisTemplate redisTemplate;@Value("${server.port}")private String serverPort;private static final String proName = "goods:1001";private final Lock lock = new ReentrantLock();private final String redisLock = "lock";@GetMapping("/by_goods")public String byGoods() {String value = UUID.randomUUID().toString() + Thread.currentThread().getName();//====setnx  加锁  设置10秒后过期Boolean flag = redisTemplate.opsForValue().setIfAbsent(redisLock, value, 10L, TimeUnit.SECONDS);if (flag == false) {return "拿锁失败";}try {String goods = redisTemplate.opsForValue().get(proName);int goodsNumber = goods == null ? 0 : Integer.parseInt(goods);if (goodsNumber > 0) {//减去之后的库存int resultNumber = goodsNumber - 1;redisTemplate.opsForValue().set(proName, String.valueOf(resultNumber));System.out.println("成功买到商品，库存还剩下：" + resultNumber + "\t服务提供端口：" + serverPort);return "成功买到商品，库存还剩下：" + resultNumber + "\t服务提供端口：" + serverPort;} else {System.out.println("商品已售完/活动结束/调用超时，欢迎下次光临" + "\t服务提供端口：" + serverPort);}return "商品已售完/活动结束/调用超时，欢迎下次光临" + "\t服务提供端口：" + serverPort;} finally {while (true) {//开启监听redisTemplate.watch(redisLock);//如果当前的锁的值等于当前线程设置锁的值那么我们就删除锁if (redisTemplate.opsForValue().get(redisLock).equals(value)) {//开始springboot对事物的支持redisTemplate.setEnableTransactionSupport(true);//开始事务redisTemplate.multi();//删除key 解锁redisTemplate.delete(redisLock);//关闭事务 如果删除失败就接着执行List<Object> exec = redisTemplate.exec();if (exec == null) {continue;}}//关闭监听redisTemplate.unwatch();break;}}}
}

来看一下运行结果比对

很明显这次没有出现超卖的现象 这是redis的解决方式

5. 使用lua脚本解决修改并且删除的原子性问题

尽管这样可以短暂的实现单机redis 的分布式事务，但是忽略了一个重要元素，如果我们面对的是集群呢？一主多从呢？有没有考虑到maser对scope传输数据时，并没有传输完，锁就返回了并且过期了，这在稳定上面有着不小的问题，基于这个，zookepor的可靠性比较高，因为有自己的分配算法，只有将自己的子节点服务分配完数据才返回，但是效率没有redis 高

基于这redis官方推荐了一种分布式锁RedLock

6.redis整合Redisson实现RedLock

注册bean

    @Beanpublic Redisson redisson() {Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379").setDatabase(0);return (Redisson)Redisson.create(config);}

先进性代码层次的修改

@RestController
public class GoodController {@Autowiredprivate StringRedisTemplate redisTemplate;@Value("${server.port}")private String serverPort;private static final String proName = "goods:1001";private final Lock lock = new ReentrantLock();private final String redisLock = "lock";@Autowiredprivate Redisson redisson;@GetMapping("/by_goods")public String byGoods() {String value = UUID.randomUUID().toString() + Thread.currentThread().getName();//获取锁名RLock redissonLock = redisson.getLock(redisLock);//设置锁redissonLock.lock();try {String goods = redisTemplate.opsForValue().get(proName);int goodsNumber = goods == null ? 0 : Integer.parseInt(goods);if (goodsNumber > 0) {//减去之后的库存int resultNumber = goodsNumber - 1;redisTemplate.opsForValue().set(proName, String.valueOf(resultNumber));System.out.println("成功买到商品，库存还剩下：" + resultNumber + "\t服务提供端口：" + serverPort);return "成功买到商品，库存还剩下：" + resultNumber + "\t服务提供端口：" + serverPort;} else {System.out.println("商品已售完/活动结束/调用超时，欢迎下次光临" + "\t服务提供端口：" + serverPort);}return "商品已售完/活动结束/调用超时，欢迎下次光临" + "\t服务提供端口：" + serverPort;} finally {if (redissonLock.isLocked()&&redissonLock.isHeldByCurrentThread()){//释放锁redissonLock.unlock();}}}
}

此时才算完全性的成功