源码url: https://github.com/zhzhair/stepsrank-spring-boot.git。

1.创建32个分表，用定时任务插入计步数据模拟用户上传步数；

2.项目启动初始化：将32个表的前200名记录插入mongodb的一个集合（表），清空后插入前200名记录，

并将第200名的步数（阈值）放到redis；

3.上传步数时，当用户的步数大于阈值时，就插入mongodb，否则不插入记录到mongodb；

4.用定时任务每隔10秒删除mongodb表中205名以后的记录；

5.用定时任务每隔1秒更新第200名的步数（阈值）到redis，同时将前200名记录放进redis的队列；

6.查询步数排名先到redis队列，查不到就去mongodb表查。

7.jmeter并发测试看查询性能。

 

程序设计简述：

技术架构：java8，spring boot2.0.0，mysql，redis，mongodb，mybatis，swagger，jmeter，idea，maven。

　　(i)添加测试数据：新建32个表，按照用户id对32取模添加测试数据到不同的表，做定时任务，每秒添加或修改300条记录。表包括user_id和步数step_count两个字段，假设手机每隔一段时间传一次累计步数，如果当日用户有记录，就修改用户的步数（增加新的步数），否则直接添加记录。部分代码如下：

@LogForTask

@Scheduled(cron = "0/1 * * * * ?")

public void uploadStep(){//定时任务每秒添加或修改300条记录

　　IntStream.range(0,300).parallel().forEach(i->stepService.uploadStep(32));

}

　　(ii)程序设计：在高并发的情况下内存是个问题(out of memory exception！)，单个mongodb文档也不能放太多的数据，所以需要设置内存不足就读取磁盘。考虑到第200名的总步数不会减少，并且越往后越“稳定”，所以把它作为阈值就可以给查询的表“瘦身”，从而避免大表排序。

　　初始化（即启动项目时）：需要将32个表的前200名都放到一个mongodb文档，再将文档前200名替换到该bson文档，同时将第200名的步数存到redis里面，部分代码如下：

@Resource

private StepService stepService;

private static StepService service;

@PostConstruct

public void init(){

　　service = this.stepService;

}

public static void main(String[] args) {

SpringApplication.run(StepsApplication.class, args);

　　//启动项目初始化排名

　　service.recordTopAll(32);

}

@Override

public void recordTopAll(int tableCount) {

　　mongoTemplate.dropCollection(StepsTop.class);//删除文档

　　IntStream.range(0,tableCount).parallel().forEach(this::insertOneTable);//将MySQL的数据插入到mongo文档

　　/*取出前200名放到list，更新mongo文档的数据为当前list的数据*/

　　Query query = new Query().with(new Sort(Sort.Direction.DESC,"totalCount")).limit(200);

　　List<StepsTop> list = mongoTemplate.find(query,StepsTop.class);

　　if(list.isEmpty()) return;

　　mongoTemplate.dropCollection(StepsTop.class);

　　mongoTemplate.insertAll(list);

　　/*redis保存阈值-第200名的步数*/

　　int size = Math.min(200,list.size());

　　redisTemplate.opsForValue().set(redisKey,String.valueOf(list.get(size - 1).getTotalCount()));

}

　　步数上传：redis的数据做定时任务更新，阈值越来越大，每次都将接收到的步数或更新后的步数与阈值比较，比这个阈值大才会去查mongo，然后对mongo文档做更新或插入操作，这个“比较”会非常频繁，但是redis“不惧怕”高并发，我们不必担心。这样就大大地减少了对mongo文档的操作，确保mongo文档数据量很少，之后查询并排序mongo文档的数据就很快了。部分代码如下：

@Override

public void uploadStep(int tableCount) {

　　int userId = new Random().nextInt(500_0000);

　　int stepCount = 1 + new Random().nextInt(5000);

　　Integer count = commonMapper.getStepCount(prefix + userId%tableCount,userId);

　　if(count != null){

　　　　commonMapper.updateSteps(prefix + userId%tableCount, userId,count + stepCount);

　　}else{

　　　　commonMapper.insertTables(prefix + userId%tableCount, userId, stepCount);

　　}

　　String tailSteps = redisTemplate.opsForValue().get(redisKey);

　　int totalCount = count == null?stepCount:count + stepCount;

　　if(tailSteps != null && totalCount > Integer.valueOf(tailSteps)){//步数超过阈值就插入或更新用户的记录

　　　　Query query = new Query(Criteria.where("userId").is(userId));

　　　　if(!mongoTemplate.exists(query,StepsTop.class)){

　　　　　　StepsTop stepsTop = new StepsTop();

　　　　　　stepsTop.setUserId(userId);

　　　　　　stepsTop.setTotalCount(stepCount);

　　　　　　mongoTemplate.insert(stepsTop);

　　　　}else{

　　　　　　System.out.println("update: " + tailSteps);

　　　　　　Update update = new Update();

　　　　　　update.set("totalStep",totalCount);

　　　　　　mongoTemplate.upsert(query,update,StepsTop.class);

　　　　}

　　}else{

　　　　StepsTop stepsTop = new StepsTop();

　　　　stepsTop.setUserId(userId);

　　　　stepsTop.setTotalCount(stepCount);

　　　　mongoTemplate.insert(stepsTop);

　　}

}

　　定时任务：每隔10秒更新一次阈值，同时删除mongo文档中200名以外的数据；每隔1秒从mongo查询排好序的前200名的数据push到redis队列，方便从redis取出排名。部分代码如下：

@Override//更新阈值，删除mongo文档中200名以外的数据

public void flushRankAll() {

　　// Query query = new Query().with(new Sort(Sort.Direction.DESC,"totalCount")).limit(201);

　　// List<StepsTop> list = mongoTemplate.find(query,StepsTop.class);//高并发场景下容易出现内存不足异常：out of memory Exception

　　TypedAggregation<StepsTop> aggregation = Aggregation.newAggregation(

　　　　StepsTop.class,

　　　　project("userId", "totalCount"),//查询用到的字段

　　　　sort(Sort.Direction.DESC,"totalCount"),

　　　　limit(200)

　　).withOptions(newAggregationOptions().allowDiskUse(true).build());//内存不足到磁盘读写，应对高并发

　　AggregationResults<StepsTop> results = mongoTemplate.aggregate(aggregation, StepsTop.class, StepsTop.class);

　　List<StepsTop> list = results.getMappedResults();

　　if(list.size() == 201){

　　　　int totalCount = list.get(199).getTotalCount();

　　　　Query query1 = new Query(Criteria.where("totalCount").lt(totalCount));

　　　　mongoTemplate.remove(query1,StepsTop.class);

　　}

}

@Override//查询排好序的前200名的数据push到redis队列

public void recordRankAll() {

　　// Query query = new Query().with(new Sort(Sort.Direction.DESC,"totalCount")).limit(200);

　　// List<StepsTop> list = mongoTemplate.find(query,StepsTop.class);

　　TypedAggregation<StepsTop> aggregation = Aggregation.newAggregation(

　　　　StepsTop.class,

　　　　project("userId", "totalCount"),//查询用到的字段

　　　　sort(Sort.Direction.DESC,"totalCount"),

　　　　limit(200)

　　).withOptions(newAggregationOptions().allowDiskUse(true).build());//内存不足到磁盘读写，应对高并发

　　AggregationResults<StepsTop> results = mongoTemplate.aggregate(aggregation, StepsTop.class, StepsTop.class);

　　List<StepsTop> list = results.getMappedResults();

　　if(list.size() == 200){

　　　　Integer stepCount = list.get(199).getTotalCount();

　　　　redisTemplate.opsForValue().set(redisKey,String.valueOf(stepCount));

　　}

　　if(!list.isEmpty()){

　　　　redisListTemplate.delete(redisQueueKey);

　　　　//noinspection unchecked

　　　　redisListTemplate.opsForList().rightPushAll(redisQueueKey,list);

　　}

}

　　查询排行榜：现在就简单了，直接到redis队列查询即可，部分代码如下：

@ApiOperation(value = "查询当日总步数排名", notes = "查询当日总步数排名")

@RequestMapping(value = "/getRankAll", method = {RequestMethod.GET}, produces = {MediaType.APPLICATION_JSON_VALUE})

public BaseResponse<List<StepsRankAllResp>> getRankAll(int begin,int pageSize) {

　　BaseResponse<List<StepsRankAllResp>> baseResponse = new BaseResponse<>();

　　List<StepsRankAllResp> list = stepService.getRankAllFromRedis(begin,pageSize);

　　if(list.isEmpty()) list = stepService.getRankAll(begin,pageSize);//redis查不到数据就从Mongo查

　　baseResponse.setCode(0);

　　baseResponse.setMsg("返回数据成功");

　　baseResponse.setData(list);

　　return baseResponse;

}

@Override//todo 从redis读取

public List<StepsRankAllResp> getRankAllFromRedis(int begin, int pageSize) {

　　List<StepsTop> stepsList = redisListTemplate.opsForList().range(redisQueueKey,begin,pageSize);

　　List<StepsRankAllResp> list = new ArrayList<>(stepsList.size());

　　for (int i = 0; i < stepsList.size(); i++) {

　　　　StepsRankAllResp stepsRankAllResp = new StepsRankAllResp();

　　　　StepsTop stepsTop = stepsList.get(i);

　　　　BeanUtils.copyProperties(stepsTop,stepsRankAllResp);

　　　　stepsRankAllResp.setRank(begin + i + 1);

　　　　list.add(stepsRankAllResp);

　　}

　　return list;

}

　　jmeter并发测试：访问接口文档--http://localhost:8080/swagger-ui.html/，调接口查询排名，配置调接口5000次，持续5秒，聚合报告如下：