redis实际应用场景及并发问题的解决

业务场景

接下来要模拟的业务场景:

每当被普通攻击的时候，有千分之三的概率掉落金币，每回合最多爆出两个金币。

1.每个回合只有15秒。

2.每次普通攻击的时间间隔是0.5s

3.这个服务是一个集群（这个要求暂时不实现）

编写接口，实现上述需求。

核心问题

可以想到要解决的主要问题是,

1.如何保证一个回合是15秒的时间?

2.如何保证如果一个回合掉落最大金币数量之后，不再掉落金币。

对于问1，我们可以选择设置回合开始的时间或者回合结束的时间，这里采用回合结束的时间。如果发现已经超过结束的时间，那么不做处理。

代码如下，second是一个回合的时间，这里就是十五秒。

    private Boolean checkRound(String id, LocalDateTime now) {if (Boolean.TRUE.equals(redisTemplate.hasKey(id))) {LocalDateTime endTime = (LocalDateTime) redisTemplate.boundValueOps(id).get();if (now.isAfter(endTime)) {log.info("该回合已经结束：回合id:{}", id);return false;}}redisTemplate.boundValueOps(id).set(now.plusSeconds(second));return true;}

对于问2，处理的方式和1一样，redis存储已经掉落的金币，若掉落金币超过最大值，则不予处理。

    private Boolean checkMoney(String id) {String moneyKey = buildMoneyKey(id);if (Boolean.TRUE.equals(stringRedisTemplate.hasKey(moneyKey))) {int money = Integer.parseInt(stringRedisTemplate.boundValueOps(moneyKey).get());if (money > maxMoney) {log.info("金钱超限。回合id:{}", id);return false;}}return true;}

如果当前回合未结束，并且掉落的金币也没有到达最大值，我们将随机生成金币返回去。

    private Boolean money(String id){Random random = new Random();int i = random.nextInt(9);if (i <= 2) {log.info("获得到了金币:{}", id);stringRedisTemplate.boundValueOps(buildMoneyKey(id)).increment();return true;}log.info("未获得到金币:{}", id);return false;}

整体代码逻辑:

@RestController
@Slf4j
public class GameController {@Value("${second:15}")private Long second;@Value("${money:2}")private Integer maxMoney;@Resourceprivate RedisTemplate redisTemplate;/*** 默认线程池*/@Resourceprivate ThreadPoolTaskExecutor threadPoolTaskExecutor;@Resourceprivate StringRedisTemplate stringRedisTemplate;@GetMapping("/attack")public Boolean attack(AttackParam attackParam) {String id = attackParam.getRoundId();log.info("攻击了一次,回合id:{}", id);LocalDateTime now = LocalDateTime.now();/**前置检查**/if (!preCheck(id, now)) {return false;}return money(id);}/*** 检测是否获得金币，获得--true ，未获得--false** @param id id* @return {@link Boolean}*/private Boolean money(String id){Random random = new Random();int i = random.nextInt(9);if (i <= 2) {log.info("获得到了金币:{}", id);stringRedisTemplate.boundValueOps(buildMoneyKey(id)).increment();return true;}log.info("未获得到金币:{}", id);return false;}private String buildMoneyKey(String id) {return "attack:money:" + id;}/*** 预检查** @param id  id* @param now 现在* @return {@link Boolean}*/private Boolean preCheck(String id, LocalDateTime now) {if (!checkRound(id, now)) {//检查回合return false;}if (!checkMoney(id)) {//检查本回合是否钱已经给够两次了return false;}return true;}/*** 校验回合是否结束** @param id id* @return {@link Boolean}*/private Boolean checkRound(String id, LocalDateTime now) {if (Boolean.TRUE.equals(redisTemplate.hasKey(id))) {LocalDateTime endTime = (LocalDateTime) redisTemplate.boundValueOps(id).get();if (now.isAfter(endTime)) {log.info("该回合已经结束：回合id:{}", id);return false;}}redisTemplate.boundValueOps(id).set(now.plusSeconds(second));return true;}/*** 校验金钱是够超限** @param id id* @return {@link Boolean}*/private Boolean checkMoney(String id) {String moneyKey = buildMoneyKey(id);if (Boolean.TRUE.equals(stringRedisTemplate.hasKey(moneyKey))) {int money = Integer.parseInt(stringRedisTemplate.boundValueOps(moneyKey).get());if (money > maxMoney) {log.info("金钱超限。回合id:{}", id);return false;}}return true;}/*** 使用线程池模拟并发测试** @return {@link String}*/@GetMapping("/test")public String test(){AttackParam attackParam = new AttackParam();attackParam.setRoundId(UUID.randomUUID().toString());for (int i = 0; i <= 10000; i++) {CompletableFuture.runAsync(() -> {this.attack(attackParam);}, threadPoolTaskExecutor);}return "aa";}
}

结果测试

接下来编写代码模拟高并发场景下是否有问题，

本次测试的并发量是1w。

    @GetMapping("/test")public String test(){AttackParam attackParam = new AttackParam();attackParam.setRoundId(UUID.randomUUID().toString());for (int i = 0; i <= 10000; i++) {CompletableFuture.runAsync(() -> {this.attack(attackParam);}, threadPoolTaskExecutor);}return "aa";}

测试结束，查询本回合掉落金币数量。

为什么我们设置的最大掉落金币数量是2，结果却是4呢?

好吧，进行第二次测试查看结果。

这一次居然是7。

说明上面这串代码在并发情况下会出现问题，即使这个并发量几十的情况依然会出问题。

问题分析

那我们就来分析一下是哪里出现了问题，出现这种原因无非就是满足写后读，那就找到读写金币的位置。

举个例子，假设线程A正在获取金币，但是这个增加的操作还没有写到redis。另外有线程B,线程C....走到了图二中查询金币数量的位置。那么这一堆线程获得仍是oldValue，这就相当于线程A的写操作是“无效的”。那么导致的结果就是金币比预期多了很多，至于多多少，取决于金币掉落的概率。

解决方案

如何解决这个问题呢?

这个问题本质上是读写分离，导致了“脏数据”。

第一个想到的也是最直接的方法肯定是加锁，但是需要考虑到这种加锁的方式只适合单体应用，如果是多个程序呢，就无法解决了。

可以将synchronized换成分布式锁。

但是加锁的方式不推荐，锁的竞争会严重影响性能。如果可以通过业务逻辑来解决，就不要去加锁。那么我们需要将读写操作放在一起，使其具有原子性。

redis中的incr操作本身就是原子的，所以我们可以将检查金币数量这个操作提前，读写放到一起。

代码如下，checkMoney就可以注掉了。

    private Boolean money(String id) {Random random = new Random();int i = random.nextInt(9);if (i <= 2) {Long increment = stringRedisTemplate.boundValueOps(buildMoneyKey(id)).increment();//将读和写放到一起 这是个原子性的if (increment > maxMoney) {log.info("金钱超限,回合{}", id);return false;}log.info("获得到了金币:{}", id);stringRedisTemplate.boundValueOps(id+"money").increment();return true;}log.info("未获得到金币:{}", id);return false;}

再次测试，可以看到数据已经是准确的了。

总结

本文讲述了redis在实际业务场景中的应用，并且看到高并发下会产生的数据错误的问题，可采取分布式锁和修改业务逻辑的方式解决，由于锁会影响到性能（请求对锁的竞争），所以更推荐后者。