业务场景
接下来要模拟的业务场景:
每当被普通攻击的时候，有千分之三的概率掉落金币，每回合最多爆出两个金币。
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;
@Resource
private RedisTemplate redisTemplate;
/**
* 默认线程池
*/
@Resource
private ThreadPoolTaskExecutor threadPoolTaskExecutor;
@Resource
private 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在实际业务场景中的应用，并且看到高并发下会产生的数据错误的问题，可采取分布式锁和修改业务逻辑的方式解决，由于锁会影响到性能（请求对锁的竞争），所以更推荐后者。