JVM第五讲：纵横数据如何应对洪峰推送-阿里云开发者社区

1、整体功能架构图

2、问题背景

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

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

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

线程池

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

异步

使用方式见这个系列

漏斗算法和令牌算法

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堆内存使用率进行限流，并不适用所有的业务场景，只是作为一个新的限流方案供大家参考扩展思路，起到一个抛砖引玉的作用

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

1、整体功能架构图

2、问题背景

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

3、数据推送

3.1、数据来源

3.2、推送流程

3.3、原限流方案

3.4、存在的问题

3.5、数据指标及工具选择

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

4、优化方案

4.1、问题原因

4.2、优化点

4.3、方案验证

5、线上案例

5.1、事件简述

5.2、事件分析

5.3、结语

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

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

1、整体功能架构图

2、问题背景

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

3、数据推送

3.1、数据来源

3.2、推送流程

3.3、原限流方案

3.4、存在的问题

3.5、数据指标及工具选择

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

4、优化方案

4.1、问题原因

4.2、优化点

4.3、方案验证

5、线上案例

5.1、事件简述

5.2、事件分析

5.3、结语

热门文章

最新文章

相关课程

相关电子书

相关实验场景