JVM第五讲：纵横数据如何应对洪峰推送

1、整体功能架构图

2、问题背景

湖南网站群的公告相关数据需要通过纵横进行数据推送，从网站群的开发了解到每天大概有10W条数据需要推送，这个数据量均摊到一天来推送并不是什么问题，但是了解到这些数据可能会集中在短时间内（10分钟）推送，此时将会出现数据洪峰的情况。

Action：出现数据洪峰时怎么应对呢？

• 应对高并发场景的三板斧就是缓存、熔断（降级）、限流，而数据推送的业务场景显然缓存和熔断并不适用，可选的就是限流了，那么常用的限流方案有哪些呢？各自的优缺点以及适用场景呢？

线程池

• 使用方式见这篇文章 JUC第二讲：深入理解 Java 中的线程池

异步

• 使用方式见这个系列

• 消息中间件第一讲：RocketMQ从入门到精通
• 消息中间件第二讲：MQ的应用beacon

漏斗算法和令牌算法

• Google guava第二讲：Ratelimiter限流原理与实现

3、数据推送

3.1、数据来源

• 业务方组装数据包发送MQ
• 纵横动态配置定时任务通过开放平台拉取数据，自行组装投递MQ

3.2、推送流程

• 纵横网关通过消费MQ，执行职责链进行数据推送

• 具体流程如下图所示
  

• 为了避免推送效率受mq消费线程数（zmq默认20）的限制或者采用线程池但不便设置具体线程池大小（每个消息体的大小差异性极大），推送时采用apache的 HttpAsyncClient（内部基于reactor模型）来执行网络请求，通过callback回调的方式来获取推送结果。

示例代码如下：

this.getHttpAsyncClient().execute(request, new FutureCallback<HttpResponse>() {
    @Override
    public void completed(HttpResponse result) {
        //todo 请求成功后的逻辑处理
    }
    @Override
    public void failed(Exception ex) {
    }
    @Override
    public void cancelled() {
    }
});

3.3、原限流方案

• 根据业务类型在当前推送队列中的数量进行限流。

• 例如公告数据的业务类型为 ANNOUNCEMENT_RELEASE_LOCAL_WEB，每种业务类型的限流数设为50，当单台实例内存中该种业务类型正在推送的数据量达到50后，该业务类型的数据从第51条开始都将会被拒绝，返给MQ稍后消费的状态，被拒绝的消息MQ会间歇性消费重试，重试时间可以动态配置。

@Component
public class BizTypeLimitHandler extends AbstractHandler<MsgInput, MsgOutput> {
    @Autowired
    FlowLimitConfig flowLimitConfig;
    @Value("${over.load.limit:50}")
    private Integer defaultLimit;
    @Override
    public void task(MsgInput input, MsgOutput output) {
        //该业务类型在当前节点的流量
        Integer flowCount = BizFlowLimitUtil.get(input.getBizType());
        //该种业务类型对应的限流
        Integer overload = this.flowLimitConfig.getBizMap().getOrDefault(input.getBizType(), this.defaultLimit);
        //overload小于0代表无限制
        if (overload >= 0 && flowCount >= overload) {
            throw new OverloadException(ErrorCode.ERROR_CODE_NSQ_LIMIT, "业务类型:" + input.getBizType() + "负载过高，阈值:" + overload + "，当前负载值:" + flowCount);
        }
    }
}

3.4、存在的问题

• 随着业务需求的增加，业务类型的种类也在不断的增加，而 所需最大资源 = 业务类型种数 * 限流数，在服务实例的资源有限的情况下，只根据业务类型的数据量来进行限流，效果会逐渐变得不理想。

3.5、数据指标及工具选择

• Arthas 是Alibaba开源的Java诊断工具，当我们遇到以下类似问题而束手无策时，Arthas可以帮助解决：

• 这个类从哪个 jar 包加载的？为什么会报各种类相关的 Exception？
• 我改的代码为什么没有执行到？难道是我没 commit？分支搞错了？
• 遇到问题无法在线上 debug，难道只能通过加日志再重新发布吗？
• 线上遇到某个用户的数据处理有问题，但线上同样无法 debug，线下无法重现！
• 是否有一个全局视角来查看系统的运行状况？
• 有什么办法可以监控到JVM的实时运行状态？
• 怎么快速定位应用的热点，生成火焰图？
• 怎样直接从JVM内查找某个类的实例？

• paas平台提供了arthas这个工具便于我们进行线上问题的排查，首先从应用发布平台进入服务实例对应的arthas控制台。
  
• arthas 提供了dashboard 命令，可以查看服务jvm的实时运行状态，如不指定刷新间隔时间，默认5s刷新一次。键入 dashboard 出现如下画面：
  

上半部分主要是当前服务jvm中的线程情况，可以看到各线程对cpu的使用率极低，基本处于闲置状态。

接着看下半部分Memory框中的信息，主要的数据项含义如下：

• heap（堆大小）
• par_eden_space(伊甸区大小)
• par_survivor_space（S区大小）
• cms_old_gen（老年代大小）
• gc.parnew.count（young gc总次数）
• gc.parnew.time（young gc总耗时）
• gc.concurrentmarksweep.count（full gc总次数）
• gc.concurrentmarksweep.time(full gc总耗时)
• 各列分别代表了已使用、总大小、最大值、已使用百分比
  

5s后如下：

结合5s前的数据，我们主要关注以下指标：

• 线程情况：线程cpu使用率并没有明显变化
• heap（堆大小）：堆使用大小增加3M
• par_eden_space(伊甸区大小)：年轻代中的伊甸区只增加3M，按照伊甸区426M的大小，大约需要710秒（约12分钟）才会触发一次young gc
• par_survivor_space（S区大小）：无变化
• cms_old_gen（老年代大小）：无变化
• gc.parnew.count（young gc总次数）：无变化
• gc.parnew.time（young gc总耗时）：无变化
• gc.concurrentmarksweep.count（full gc总次数）：无变化
• gc.concurrentmarksweep.time(full gc总耗时)：无变化

这是服务无流量基本处于闲置状态时一个情况，接下来模拟积压数据进行推送时的场景，数据由测试同学提前通过自动化脚本投递到MQ当中。

积压5000条数据

• 使用arthas命令 dashboard -i 1000，按照1s的间隔输出：
  1S后：
  

对比两次数据发现：

• 线程情况：zmq默认的20个消费者线程都处于活跃状态占用cpu资源
• heap（堆大小）：已使用大小从293M上升至478M
• par_eden_space(伊甸区大小)：发生young gc之前伊甸区使用23m，伊甸区总大小为426m，发生young gc之后伊甸区使用了211m，这说明在1s之内至少增加了（426-23）+211 = 614 m大小的对象
• par_survivor_space（S区大小）：young gc之前s区大小为31m，young gc之后s区大小为29m
• cms_old_gen（老年代大小）：无变化
• gc.parnew.count（young gc总次数）：发生了1次young gc
• gc.parnew.time（young gc总耗时）：时间增加了（9018-8992）= 26 毫秒，为一次young gc的时长
• gc.concurrentmarksweep.count（full gc总次数）：无变化
• gc.concurrentmarksweep.time(full gc总耗时)：无变化

1s一次young gc，可能是时间间隔太短老年代的数据没有变化，我们按照5s的间隔，再观察一下老年代：

5s后：

对比两次数据，关键信息如下：

• 5s之内发生了7次young gc
• 老年代由233m增长至265m，增长了32m左右，按照老年代512m的大小，大约80s就会发生一次full gc

积压1W条数据

开始：

1S后：

对比两次数据得知：

1s之内发生了两次young gc

同时老年代从304m增长至314m，1s增长了10m，老年代大小为512m，按照这个速率，大约50s就会触发一次full gc

此时下游还未将数据推送完成，同时上游还不断的从mq中消费消息，继续观察

1s后：

对比两次数据发现：

• gc线程的cpu使用率居高不下
• 发生了一次full gc

一秒前老年代已使用大小为418M，总大小为512M，1s后发现触发了一次full gc，按照上面分析出老年代1s增加10M的速度，老年代的剩余空间是足够的，为什么还会出现full gc这种情况呢？

full gc发生的时机有哪些呢？

• 第一种情况：老年代可用内存小于年轻代全部存活对象大小，同时没有开启空间担保参数（-XX:-HandlePromotionFailure）

• 从jdk6之后，HandlePromotionFailure 参数不会再影响到虚拟机的空间分配担保策略，我们使用的都是jdk8，所以第一种情况不满足

• 第二种情况：老年代可用内存小于年轻代全部对象大小，开启了空间担保参数，但是可用内存小于历次年轻代GC后进入老年代的平均对象大小

• 根据之前的分析，1s内进入老年代的对象大小大约为10M，而目前老年代剩余大小约为（512-418）= 94M，所以第二种情况也不满足

• 第三种情况：年轻代Minor GC后存活对象大于Survivor，就会进入老年代，但是老年代内存不足

• 同第二种情况，第三种情况也不太满足

• 第四种情况：设置了参数-XX:CMSInitiatingOccupancyFaction，老年代可用内存大于历次年轻代GC后进入老年代的对象的平均大小，但是老年代已使用内存超过该参数指定的比例，自动触发Full GC（建议92%）

• 发现jvm启动参数中加了该参数： -XX:CMSInitiatingOccupancyFraction = 80，该参数表示老年代在达到 512 * 80% = 409m 大小时就会触发一次full gc。该参数主要是为了解决cmf（Concurrent Mode Failure）问题，不过该参数也会导致full gc更加的频繁。
• 按照这个速率，老年代从目前的305m增长至409m只需10s，意味着10s后就会发生一次full gc！

• 现在我们知道了提前触发了full gc的原因是由于 CMSInitiatingOccupancyFraction 参数的配置，正常情况下设为80%也不会有什么问题，但是有没有这种极端情况呢：
  

发生full gc后老年代的空间并没有回收多少，老年代已使用空间大小一直在设定的阈值之上，导致不停的full gc？

积压1.5W条数据

开始：

5S后：

对比两次数据，得知以下关键信息：

• 线程情况：cms垃圾回收线程cpu占用率高达42%+
• full gc次数：5s内发生了3次full gc
• full gc总耗时：总耗时由14742 增加至 15131

5s之内居然发生了3次full gc，平均每次full gc耗时（15131-14742）/ 3 = 129 ms，虽然出现了full gc频繁的情况，但是还不至于造成服务不可用。

积压2W条数据

开始:

5s后：

对比两次数据发现：

• 线程情况：cms垃圾回收线程cpu占用率极高
• 老年代：已使用511m（总大小512m）
• full gc次数：5s内发生了3次full gc
• full gc总耗时：总耗时由14742 增加至 15131
• 5s内发生了3次full gc，老年代始终处于已使用511m（总大小512m）的情况,每次的full gc平均耗时 （81942-79740）/3=734 ms，相当于5s内有2.2s的时间都在full gc。

此时查看日志发现数据推送时发生大量的socket连接超时：

再查看下当时的gc日志

使用默认的jvm参数，gc日志放置在该路径下：/opt/zcy/modules/gc.log

查看gc日志发现：两次full gc之间只差了1.4s左右，从524287k回收至524275k，只回收了12k的内存空间，却花费了0.71s时间，系统有一半的时间都在进行full gc，造成系统不可用。

使用监控大盘（grafana）查看下当时的cpu、网络io情况

通过监控大盘可以看到由于full gc频繁，导致cpu被打满，网络请求相关的线程得不到有效的调度，导致网络io吞吐下降。

4、优化方案

4.1、问题原因

• 通过上面测试可以发现系统存在的问题主要是由于下游消费速率（网络请求）跟不上上游的投递速率（mq消费），导致内存逐渐被打满，系统频繁full gc造成服务不可用，直至产生OOM程序崩溃。

4.2、优化点

• 在该场景中系统的主要瓶颈在于jvm内存大小上面，避免系统频繁full gc即可达到我们的目的，所以可以尝试基于jvm堆内存使用率来对上游进行限流控制。

jvm参数优化

• 原来年轻代大小设为1024M并不适用于纵横这种数据推送场景。通过观察发现老年代常驻对象150M左右，考虑浮动垃圾等，老年代分配512M大小，所以年轻代调整为1.5g大小：-Xmn1536M
• 修改年轻代中的伊甸区和s区的比例由10:1:1修改为8:1:1，避免young gc后存活对象过多直接进入老年代： -XX:SurvivorRatio=8
• 元空间大小分配256m。-XX:MaxMetaspaceSize=256M -XX:MetaspaceSize=256M
• 线程栈最大1m：-Xss1M
• 针对年轻代使用ParNew垃圾收集器：-XX:+UseParNewGC

所以最终优化后的jvm参数如下：

-Xmx2g -Xms2g -Xmn1536M -Xss1M -XX:MaxMetaspaceSize=256M -XX:MetaspaceSize=256M -XX:SurvivorRatio=8

-XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSMaxAbortablePrecleanTime=5000 -XX:+CMSClassUnloadingEnabled -XX:CMSInitiatingOccupancyFraction=80 -XX:+UseCMSInitiatingOccupancyOnly -XX:+ExplicitGCInvokesConcurrent -Xloggc:/opt/zcy/modules/gc.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/opt/zcy/modules/java.hprof

增加资源限流

• 纵横推送流程是基于责任链的模式来进行设计的，所以只需在链中再增加一个限流器即可。

• 限流器中可设定一个jvm堆内存的使用率，当超过这个阈值后对当前的消费线程进行阻塞，直到使用率低于阈值后再进行放行
• 当触发限流阻塞消费者线程时，为应对特殊情况，可设置一个最大阻塞时间，超过该时间后放行该消费者线程

其中阈值的配置范围给出一个参考公式：

• 最大阈值百分比 = （年轻代可使用最大值 + 老年代可使用最大值 * -XX:CMSInitiatingOccupancyFraction参数值）/ 堆可使用最大值
• 最小阈值百分比= 年轻代可使用最大值 / 堆可使用最大值

代入相关参数，即

• 最大阈值百分比 = （1381 + 512*0.8 ）/ 1894 = 94%
• 最小阈值百分比 = 1381 / 1894 = 72%

参考代码如下：

@Slf4j
@Component
public class ResourceLimitHandler extends AbstractHandler<MsgInput, MsgOutput> {
    /**
     * 是否启用资源限流
     */
    @Value("${flow.limit.heap.trigger:true}")
    private Boolean trigger;
    /**
     * jvm堆限流阈值
     */
    @Value("${flow.limit.heap.threshold:70}")
    private Integer threshold;
    /**
     * 单次睡眠时间(毫秒)
     */
    @Value("${flow.limit.heap.sleepTime:1000}")
    private Integer sleepTime;
    /**
     * 最大阻塞时间(毫秒)，-1代表无限制
     */
    @Value("${flow.limit.heap.maxBlockTime:15000}")
    private Integer maxBlockTime;
    private MemoryMXBean memoryMXBean = ManagementFactory.getMemoryMXBean();
    @SneakyThrows
    @Override
    public void task(MsgInput input, MsgOutput output) {
        if (!this.trigger) {
            return;
        }
        boolean isLimit = false;
        long startTime = System.currentTimeMillis();
        double percent = this.getHeapUsedPercent();
        while (this.trigger && percent >= this.threshold ) {
            if (!isLimit) {
                isLimit = true;
                log.warn("资源使用过高，开启资源限流。heap:【{}】,used:【{}】,percent:【{}%】,threshold:【{}%】,msgId:【{}】,bizType:【{}】,bizId:【{}】", this.getHeapMax(), this.getHeapUsed(), percent, this.threshold, input.getMessageId(), input.getBizType(), input.getBusinessId());
            }
            if (this.maxBlockTime >= 0 && (System.currentTimeMillis() - startTime) > this.maxBlockTime) {
                log.warn("资源使用过高,但超过最大阻塞时间({}毫秒)，开始放行。heap:【{}】,used:【{}】,percent:【{}%】,threshold:【{}%】,msgId:【{}】,bizType:【{}】,bizId:【{}】", this.maxBlockTime, this.getHeapMax(), this.getHeapUsed(), percent, this.threshold, input.getMessageId(), input.getBizType(), input.getBusinessId());
                //兜底，防止因为限流导致年轻代无新对象产生，达不到触发young gc的条件，所以手动触发一次full gc
                synchronized (ResourceLimitHandler.class) {
                    if ((percent = this.getHeapUsedPercent()) >= this.threshold) {
                        System.gc();
                        log.warn("内存空间不足，手动触发full gc。heap:【{}】,used:【{}】,percent:【{}%】,threshold:【{}%】,msgId:【{}】,bizType:【{}】,bizId:【{}】", this.getHeapMax(), this.getHeapUsed(), percent, this.threshold, input.getMessageId(), input.getBizType(), input.getBusinessId());
                    }
                }
                return;
            }
            TimeUnit.MILLISECONDS.sleep(this.sleepTime);
            percent = this.getHeapUsedPercent();
        }
        if (isLimit) {
            log.info("资源使用率恢复，开始放行，限流时长：【{}】。heap:【{}】,used:【{}】,percent:【{}%】,threshold:【{}%】,msgId:【{}】,bizType:【{}】,bizId:【{}】", System.currentTimeMillis() - startTime, this.getHeapMax(), this.getHeapUsed(), percent, this.threshold, input.getMessageId(), input.getBizType(), input.getBusinessId());
        }
    }
    /**
     * 计算堆的使用百分比
     * @return
     */
    private double getHeapUsedPercent() {
        long max = this.getHeapMax();
        long used = this.getHeapUsed();
        double percent = NumberUtil.div(used, max) * 100;
        return percent;
    }
    /**
     * 可用堆最大值
     * @return
     */
    private long getHeapMax() {
        MemoryUsage memoryUsage = this.memoryMXBean.getHeapMemoryUsage();
        return memoryUsage.getMax();
    }
    /**
     * 已使用堆大小
     * @return
     */
    private long getHeapUsed() {
        MemoryUsage memoryUsage = this.memoryMXBean.getHeapMemoryUsage();
        return memoryUsage.getUsed();
    }
}

4.3、方案验证

资源配置

• 数据量：5w条积压数据
• 实例配置：1核4g
• 实例数：2

资源限流前

推送完成后：

对应的监控大盘：

• 现象：由于内存被打满，jvm频繁进行full gc，导致cpu被打满，数据推送线程获取的cpu时间片减少，引起网络io下降，大量请求返回socket连接超时。

资源限流后

• 在对jvm参数进行优化以及添加资源限流器后，再次模拟数据积压情况观察效果。

推送完成后：

对应的监控大盘情况：

结果比对

• 对比优化前后，使用新的限流方案后的效果还是挺显著的，效能近乎提升了一倍

5、线上案例

5.1、事件简述

• 3月17日晚7点左右收到告警通知：安徽环境-纵横内存使用率过高-超过95%事件

处理过程

• 登录arthas控制台，发现两台配置为2g内存的服务实例jvm内存使用率高达99%，在观察过程中一台实例挂掉自动重启。
  
• 查看应用日志以及Apollo配置确认应用的资源限流功能已经触发，正常数据量推送情况不可能还会出现内存被打满的情况，必定是出现了数据洪峰！。
  
• 数据库查询从中午12点至今的数据推送量发现达到了100W+，其中网站群公告、竞争性谈判公告各占近50%，该数据量相当于安徽环境6个月推送的数据总量。
  
• 联系安徽陆安，经沟通发现是对方发版有误导致历史数据全部重推了一遍，相当于一下午推送了历史6个月的数据。
  
• 立即申请运维同学紧急协助升配扩容为8G内存 4实例，纵横Apollo调整资源限流参数，同时联系安徽陆安停止历史重复数据推送。
  

5.2、事件分析

• 上游业务方未对业务数据进行判断导致压力转嫁至纵横网关侧，造成数据洪峰。
  
• 纵横机器实例内存配置不足以支撑该数据量，触发了资源限流，但为何仍然出现了内存使用高达99%的情况？

• 当时Apollo中配置的限流阈值为 80%，单次限流最大阻塞时长为10秒，之所以触发了资源限流内存还被打满的原因是阻塞10s后，由于下游第三方接口响应较慢，之前推送的数据占据的内存资源仍未被释放掉，进而导致内存逐渐被打满，所以针对该种特殊情况，也需要合理配置单次限流最大阻塞时长。

5.3、结语

• 每种方案都有各自的优缺点、局限性，根据jvm堆内存使用率进行限流，并不适用所有的业务场景，只是作为一个新的限流方案供大家参考扩展思路，起到一个抛砖引玉的作用