java - Java 进程消耗超过 100% 的 CPU

Question

我在具有 Ubuntu 18.04.2 的 VM 上运行 spring-boot 应用程序。该应用程序基本上具有 REST API，它从数据库中获取数据，对其进行处理并发送回客户端。运行一段时间后，应用程序的 CPU 消耗达到 190% 左右。我使用 top 命令检查了这个。当我执行 top -H 时，我可以看到 2 个线程，每个线程都消耗大约 90% 的 CPU，如下所示。

PID USER     PR NI  VIRT    RES    SHR S %CPU %MEM     TIME+ COMMAND   
2185  root   20 0 3680020 626440 17968 R 99.0  7.7  23:55.98 java                                                                                                       
28726 root   20 0 3680020 626440 17968 R 96.4  7.7  24:01.21 java

在此之后，应用程序将停止工作。我必须杀死并重新启动应用程序才能使其再次工作。

为了调试这个问题，我尝试从 postman runner 中使用 0 秒延迟、20 次迭代和 5 个并行请求的 API。我可以通过这样做来重现这个问题。我使用 jstack -F 进行了线程转储，并观察到消耗 CPU 的线程处于 IN_JAVA 状态。请检查以下内容。

Thread 2185: (state = IN_JAVA)
 - java.util.TreeMap.getEntry(java.lang.Object) @bci=79, line=359 (Compiled frame; information may be imprecise)
 - java.util.TreeMap.get(java.lang.Object) @bci=2, line=278 (Compiled frame)
 - services.impl.ChannelServiceImpl.currentSalesCategoryParam1(java.util.Map, java.lang.String, java.lang.String, java.lang.String, java.lang.String, java.lang.Double, java.lang.Double) @bci=154, line=206 (Compiled frame)
 - services.impl.ChannelServiceImpl.lambda$getChannelsVoFromSalesRecord$2(java.util.Map, models.SalesRecord) @bci=65, line=103 (Compiled frame)
 - services.impl.ChannelServiceImpl$$Lambda$13.accept(java.lang.Object) @bci=8 (Compiled frame)
 - java.util.stream.ForEachOps$ForEachOp$OfRef.accept(java.lang.Object) @bci=5, line=184 (Compiled frame)
 - java.util.ArrayList$ArrayListSpliterator.forEachRemaining(java.util.function.Consumer) @bci=99, line=1382 (Compiled frame)
 - java.util.stream.AbstractPipeline.copyInto(java.util.stream.Sink, java.util.Spliterator) @bci=32, line=481 (Compiled frame)
 - java.util.stream.ForEachOps$ForEachTask.compute() @bci=103, line=291 (Compiled frame)
 - java.util.concurrent.CountedCompleter.exec() @bci=1, line=731 (Compiled frame)
 - java.util.concurrent.ForkJoinTask.doExec() @bci=10, line=289 (Interpreted frame)
 - java.util.concurrent.ForkJoinPool$WorkQueue.runTask(java.util.concurrent.ForkJoinTask) @bci=21, line=1056 (Interpreted frame)
 - java.util.concurrent.ForkJoinPool.runWorker(java.util.concurrent.ForkJoinPool$WorkQueue) @bci=35, line=1692 (Interpreted frame)
 - java.util.concurrent.ForkJoinWorkerThread.run() @bci=24, line=157 (Interpreted frame)

根据我最初的理解，我认为并行流是罪魁祸首，但我不确定。

我正在使用以下代码来处理大约 9500 条记录。currentSalesCategoryParam1 为进一步的数据按摩创建了一个 5 级嵌套地图。

        salesRecords.parallelStream().forEach(record -> {
            String levelOneKey = MappingUtils.resolveBaseChannel(record.getBuyingGroupDesc(),
                    record.getRetailChannel());
            String levelTwoKey = MappingUtils.resolveChannelName(record.getBuyingGroupDesc(),
                    record.getRetailChannel());
            String levelThreeKey = record.getBrandName();
            String levelFourKey = MappingUtils.resolveSalesType(record.getLyMeasure());
            Double currentRowVariance = MappingUtils.currentRowVariance(record.getTyValue(), record.getLyValue());
            currentSalesCategoryParam1(dataAggregationMap, levelOneKey, levelTwoKey, levelThreeKey, levelFourKey,
                    currentRowVariance, record.getLyValue());
});


private static void currentSalesCategoryParam1(
            Map<String, Map<String, Map<String, Map<String, Map<String, Double>>>>> dataAggregation, String levelOneKey,
            String levelTwoKey, String levelThreeKey, String levelFourKey, Double variance, Double lySum) {

        Map<String, Map<String, Map<String, Map<String, Double>>>> levelOneMap = dataAggregation.get(levelOneKey);
        if (CollectionUtils.isEmpty(levelOneMap)) {
            levelOneMap = new TreeMap<>();
        }

        Map<String, Map<String, Map<String, Double>>> levelTwoMap = levelOneMap.get(levelTwoKey);
        Map<String, Map<String, Map<String, Double>>> levelTwoTotalMap = levelOneMap.get("Total");
        if (CollectionUtils.isEmpty(levelTwoMap)) {
            levelTwoMap = new TreeMap<>();
        }
        if (CollectionUtils.isEmpty(levelTwoTotalMap)) {
            levelTwoTotalMap = new TreeMap<>();
        }

        Map<String, Map<String, Double>> levelThreeMap = levelTwoMap.get(levelThreeKey);
        Map<String, Map<String, Double>> levelThreeTotalMap = levelTwoTotalMap.get(levelThreeKey);
        if (CollectionUtils.isEmpty(levelThreeMap)) {
            levelThreeMap = new TreeMap<>();
        }
        if (CollectionUtils.isEmpty(levelThreeTotalMap)) {
            levelThreeTotalMap = new TreeMap<>();
        }

        Map<String, Double> levelFourMap = levelThreeMap.get(levelFourKey);
        Map<String, Double> levelFourTotalMap = levelThreeTotalMap.get(levelFourKey);
        if (CollectionUtils.isEmpty(levelFourMap)) {
            levelFourMap = new TreeMap<>();
            levelFourMap.put(MappingConstants.VARIANCE, 0.0);
            levelFourMap.put(MappingConstants.LY_SUM, 0.0);
        }
        if (CollectionUtils.isEmpty(levelFourTotalMap)) {
            levelFourTotalMap = new TreeMap<>();
            levelFourTotalMap.put(MappingConstants.VARIANCE, 0.0);
            levelFourTotalMap.put(MappingConstants.LY_SUM, 0.0);
        }

        levelFourMap.put(MappingConstants.VARIANCE, levelFourMap.get(MappingConstants.VARIANCE) + variance);
        levelFourMap.put(MappingConstants.LY_SUM, levelFourMap.get(MappingConstants.LY_SUM) + lySum);
        levelFourTotalMap.put(MappingConstants.VARIANCE, levelFourMap.get(MappingConstants.VARIANCE) + variance);
        levelFourTotalMap.put(MappingConstants.LY_SUM, levelFourMap.get(MappingConstants.LY_SUM) + lySum);
        levelThreeMap.put(levelFourKey, levelFourMap);
        levelThreeTotalMap.put(levelFourKey, levelFourTotalMap);
        levelTwoMap.put(levelThreeKey, levelThreeMap);
        levelTwoTotalMap.put(levelThreeKey, levelThreeTotalMap);
        levelOneMap.put(levelTwoKey, levelTwoMap);
        levelOneMap.put("Total", levelTwoTotalMap);
        dataAggregation.put(levelOneKey, levelOneMap);
    }```

标签： javarestspring-bootparallel-processingstream