你好呀，我是歪歪。
是这样的，前几天我收到掘金的系统通知说我在活动中中奖了。
然后我点进去一看，好家伙荣登榜眼：
是我之前写的这篇文章：《面试官：要不我们聊一下“心跳”的设计？》
但是这不重要，重要的是我看一下第一名写的文章，有点东西，又学到一个问题的解决方案，所以我想分享一下。
文章地址在这里：
https://juejin.cn/post/7041076758711369764
高并发 PV 问题
他的文章标题是这样的：
首先他给出了一个业务场景：在一些需要统计 PV(Page View), 即页面浏览量或点击量高并发系统中，如：知乎文章浏览量，淘宝商品页浏览量等，需要统计相应数据做分析。
比如我的公众号阅读量大概在 3000 左右，如果要统计我这种小号主的 PV 其实就很简单，就用 Redis 的 incr 命令轻轻松松就实现了。
但是，假设微信公众号每天要统计 10 万篇文章，每篇文章的访问量 10 万，如果采用 Redis 的 incr命令来实现计数器的话，每天 Redis=100 亿次的写操作，按照每天高峰 12 小时来算，那么 Redis 大约 QPS=57万。
如此大的并发量，CPU 肯定满负载运行，网络资源消耗也巨大，所以直接使用 incr 命令这种技术方案是行不通的。
假设这是一个面试场景题，你会怎么去回答呢？
其实你也别想的有多复杂，剥离开场景，这无外乎就是一个高并发的问题。
而高并发问题的解决方案，基本上逃不过这三板斧：缓存、拆分、加钱。
所以这个老哥给出的方案就是：缓存。
二级缓存
Redis 都已经是缓存了，那么再加缓存算什么回事呢？
那就算是二级缓存了。
而且这个缓存，就在 JVM 内存里面，比 Redis 还快。
其核心思想是减少 Redis 的访问量。这些理论的东西，大家应该都知道。
那么通过什么方案去减少 Redis 的访问量呢，这个二级缓存应该怎么去设计呢？
首先，文章服务采用了集群部署，在线上可以部署多台。
然后每个文章服务，增加一级 JVM 缓存，即用 Map 存储在 JVM，key 为当前请求所属的时间块。
就是这个意思：
Map<Long,Map<Integer,Integer>> = Map<时间块，Map<文章id,访问量>>
但是我觉得巧妙的地方在于这里提到的“时间块”的概念。
什么是时间块?
就是把时间切割为一块块，例如：一篇文章在1小时，30分钟、5分钟的时间内产生了多少阅读量。
如何切割时间块呢？
这里利用到了时间是不断增长的特性。
时间戳是自 1970 年1月1日（00:00:00 GMT）至当前时间的总数，通过确定时间块大小，算出当前请求所属的时间戳从 1970 年算起位于第几个时间块，这个算出来的第几个时间块就是小时 key ，即 map 的 key。
举个例子：
我们把时间按照“小时”的维度进行划分。
先把当前的时间转换为为毫秒的时间戳，然后除以一小时，即当前时间 T/1000*60*60=小时key，然后用这个小时序号作为key。
比如：
2021-12-26 15:00:00 = 1640502000000毫秒
那么小时key= 1640502000000/1000*60*60=455695，即是距离 1970 年开始算的第 455695 个时间块。
2021-12-26 15:10:00 = 1640502600000毫秒，那么算出来的 key =455695.1，向下取整，key 还是等于 455695。
意思是这段时间的时间块是一样的，所以统计到 JVM 内存中的 Map 的时候，对应的 key 是一样的。
画个图示意一下：
上图中，在 2021-12-25 15:00:00 到 2021-12-25 15:59:59 时间段内产生的阅读量都会映射到 Map 的 key= 455695 的位置去。
在 2021-12-25 16:00:00 到 2021-12-25 16:59:59 时间段内产生的阅读量都会映射到 Map 的 key= 455696 的位置去。
以此类推，每一次PV操作时，先计算当前时间是那个时间块，然后存储Map中。
整体方案
当我们把数据缓存到内存中之后，就极大的减少了对于 Redis 的访问。
但是我们还是得把数据同步到 Redis 里面去，因为访问文章数据的时候还是得从 Redis 中获取数据。
所以，这里就涉及到一个问题：什么时候、怎么把数据同步到 Redis 呢？
看一下作者给出的方案设计：
整体流程还是比较清楚，主要说一下里面的两个定时任务。
其中一级缓存定时器的逻辑是这样的：假设每 5 分钟（可以根据需求调整）从 JVM 的 Map 里面把时间块的阅读 PV 读取出来，然后 push 到 Redis 的 list 数据结构中。
list 存储的数据为 Map<文章Id，访问量PV>，即每个时间块的 PV 数据。
另外一个二级缓存定时器的逻辑是这样的：每 6 分钟（需要比一级缓存的时间长），从 Redis 的 list 数据结构中 pop 出数据，即Map<文章Id，访问量PV>。
然后把对应的数据同步到 DB 和 Redis 中。
代码实战
代码主要分为四个步骤，我也把代码粘过来给大家看看。
步骤1：PV请求处理逻辑
代码语言：javascript复制//保存时间块和pv数据的map
public static final Map<Long, Map<Integer,Integer>> PV_MAP=new ConcurrentHashMap();
/**
* pv请求调用：
* 即当前时间T/1000*60*60=小时key，然后用这个小时序号作为key。
* 例如：
* 2021-11-09 15:30:00 = 1636443000000毫秒
* 小时key=1636443000000/1000*60*60=454567.5=454567
*
* 每一次PV操作时，先计算当前时间是那个时间块，然后存储Map中。
* @param id 文章id
*/
public void addPV(Integer id) {
//生成环境：时间块为5分钟
//为了方便测试 改为1分钟 时间块
int timer=1;
long m1=System.currentTimeMillis()/(1000*60*timer);
//拿出这个时间块的所有文章数据
Map<Integer,Integer> mMap=Constants.PV_MAP.get(m1);
if (CollectionUtils.isEmpty(mMap)){
mMap=new ConcurrentHashMap();
mMap.put(id,new Integer(1));
//<1分钟的时间块，Map<文章Id,访问量>>
Constants.PV_MAP.put(m1, mMap);
}else {
//通过文章id 取出浏览量
Integer value=mMap.get(id);
if (value==null){
mMap.put(id,new Integer(1));
}else{
mMap.put(id,value 1);
}
}
}
步骤2：一级缓存定时器消费
定时（5分钟)从 JVM 的 Map 把时间块的阅读 PV 取出来，然后 push 到 Reids 的 list 数据结构中，list 的存储的数据为 Map<文章id，访问量PV> 即每个时间块的 PV 数据
代码语言：javascript复制/**
* 一级缓存定时器消费调用方法：
* 定时器，定时（5分钟）从jvm的map把时间块的阅读pv取出来，
* 然后push到reids的list数据结构中，list的存储的书为Map<文章id，访问量PV>即每个时间块的pv数据
*/
public void consumePV(){
//为了方便测试 改为1分钟 时间块
long m1=System.currentTimeMillis()/(1000*60*1);
Iterator<Long> iterator= Constants.PV_MAP.keySet().iterator();
while (iterator.hasNext()){
//取出map的时间块
Long key=iterator.next();
//小于当前的分钟时间块key ，就消费
if (key<m1){
//先push
Map<Integer,Integer> map=Constants.PV_MAP.get(key);
//push到reids的list数据结构中，list的存储的书为Map<文章id，访问量PV>即每个时间块的pv数据
this.redisTemplate.opsForList().leftPush(Constants.CACHE_PV_LIST,map);
//后remove
Constants.PV_MAP.remove(key);
log.info("push进{}",map);
}
}
}
步骤3：二级缓存定时器消费
定时(5分钟)，从 Redis 的 list 数据结构 pop 弹出 Map<文章id，访问量PV>，弹出来做了2件事:
- 先把 Map<文章id，访问量PV>，保存到数据库
- 再把 Map<文章id，访问量PV>，同步到 Redis 缓存的计数器 incr
步骤4：查看浏览量
用了一级缓存，所有的高并发流量都收集到了本地 JVM，然后 5 分钟同步给二级缓存，从而给 Redis 降压。
代码语言：javascript复制@GetMapping(value = "/view")
public String view(Integer id) {
//文章pv的key
String key= Constants.CACHE_ARTICLE id;
//调用redis的get命令
String n=this.stringRedisTemplate.opsForValue().get(key);
log.info("key={},阅读量为{}",key, n);
return n;
}
对应视频
另外，我在 B 站也找到这篇文章对应的视频：
https://www.bilibili.com/video/BV1PY411p7MG?p=1
如果大家有没有看明白的地方，可以去 B 站看一下对应的视频，讲的还是很清楚的。
整体方案是没有问题的，时间块的设计也非常的巧妙。
当然了如果你非要找方案的瑕疵的话，那就是数据时效性和数据一致性的问题了。
其实我了解到这个方案之后，我还是觉得万变不离其宗，这个方案就是一种合并提交的理念。
