慧銷平臺ThreadPoolExecutor記憶體洩漏分析

語言: CN / TW / HK

作者:京東零售 馮曉濤

問題背景

京東生旅平臺慧銷系統,作為平臺系統對接了多條業務線,主要進行各個業務線廣告,召回等活動相關內容與能力管理。

最近根據告警發現記憶體持續升高,每隔2-3天會收到記憶體超過閾值告警,猜測可能存在記憶體洩漏的情況,然後進行排查。根據24小時時間段記憶體監控可以發現,容器的記憶體在持續上升:

問題排查

初步估計記憶體洩漏,檢視24小時時間段jvm記憶體監控,排查jvm記憶體回收情況:

YoungGC和FullGC情況:

通過jvm記憶體分析和YoungGC與FullGC執行情況,可以判斷可能原因如下:

1、 存在YoungGC但是沒有出現FullGC,可能是物件進入老年代但是沒有到達FullGC閾值,所以沒有觸發FullGC,物件一直存在老年代無法回收

2、 存在記憶體洩漏,雖然執行了YoungGC,但是這部分記憶體無法被回收

通過執行緒數監控,觀察當前執行緒情況,發現當前執行緒數7427個,並且還在不斷上升,基本判斷存在記憶體洩漏,並且和執行緒池的不當使用有關:

通過JStack,獲取執行緒堆疊檔案並進行分析,排查為什麼會有這麼多執行緒:

發現通過執行緒池建立的執行緒數達7000+:

程式碼分析

分析程式碼中ThreadPoolExecutor的使用場景,發現在一個worker公共類中定義了一個執行緒池,worker執行時會使用執行緒池進行非同步執行。

 public class BackgroundWorker {
 
     private static ThreadPoolExecutor threadPoolExecutor;
 
     static {
         init(15);
     }
 
     public static void init() {
         init(15);
     }
 
     public static void init(int poolSize) {
         threadPoolExecutor =
                 new ThreadPoolExecutor(3, poolSize, 1000, TimeUnit.MINUTES, new LinkedBlockingDeque<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());
     }
 
     public static void shutdown() {
         if (threadPoolExecutor != null && !threadPoolExecutor.isShutdown()) {
             threadPoolExecutor.shutdownNow();
         }
     }
 
     public static void submit(final Runnable task) {
         if (task == null) {
             return;
         }
         threadPoolExecutor.execute(() -> {
             try {
                 task.run();
             } catch (Exception e) {
                 e.printStackTrace();
             }
         });
     }
 
 }

廣告快取重新整理worker使用執行緒池的程式碼:

 public class AdActivitySyncJob {

    @Scheduled(cron = "0 0/5 * * * ?")
    public void execute() {
        log.info("AdActivitySyncJob start");
        List<DicDTO> locationList = locationService.selectLocation();
        if (CollectionUtils.isEmpty(locationList)) {
            return;
        }

        //中間省略部分無關程式碼

        BackgroundWorker.init(40);
        locationCodes.forEach(locationCode -> {
            showChannelMap.forEach((key,value)->{
                BackgroundWorker.submit(new Runnable() {
                    @Override
                    public void run() {
                        log.info("AdActivitySyncJob,locationCode:{},showChannel:{}",locationCode,value);
                        Result<AdActivityDTO> result = notLoginAdActivityOuterService.getAdActivityByLocationInner(locationCode, ImmutableMap.of("showChannel", value));
                        LocalCache.AD_ACTIVITY_CACHE.put(locationCode.concat("_").concat(value), result);
                    }
                });
            });
        });
        log.info("AdActivitySyncJob end");
    }

    @PostConstruct
    public void init() {
        execute();
    }
}

原因分析:猜測是worker每次執行,都會執行init方法,建立新的執行緒池,但是區域性建立的執行緒池並沒有被關閉,導致記憶體中的執行緒池越來越多,ThreadPoolExecutor在使用完成後,如果不手動關閉,無法被GC回收。

分析驗證

驗證區域性執行緒池ThreadPoolExecutor建立後,如果不手動關閉,是否會被GC回收:

public class Test {
    private static ThreadPoolExecutor threadPoolExecutor;

    public static void main(String[] args) {
        for (int i=1;i<100;i++){
            //每次均初始化執行緒池
            threadPoolExecutor =
                    new ThreadPoolExecutor(3, 15, 1000, TimeUnit.MINUTES, new LinkedBlockingDeque<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());

            //使用執行緒池執行任務
            for(int j=0;j<10;j++){
                submit(new Runnable() {
                    @Override
                    public void run() {
                    }
                });
            }

        }
        //獲取當前所有執行緒
        ThreadGroup group = Thread.currentThread().getThreadGroup();
        ThreadGroup topGroup = group;
        // 遍歷執行緒組樹,獲取根執行緒組
        while (group != null) {
            topGroup = group;
            group = group.getParent();
        }
        int slackSize = topGroup.activeCount() * 2;
        Thread[] slackThreads = new Thread[slackSize];
        // 獲取根執行緒組下的所有執行緒,返回的actualSize便是最終的執行緒數
        int actualSize = topGroup.enumerate(slackThreads);
        Thread[] atualThreads = new Thread[actualSize];
        System.arraycopy(slackThreads, 0, atualThreads, 0, actualSize);
        System.out.println("Threads size is " + atualThreads.length);
        for (Thread thread : atualThreads) {
            System.out.println("Thread name : " + thread.getName());
        }
    }

    public static void submit(final Runnable task) {
        if (task == null) {
            return;
        }
        threadPoolExecutor.execute(() -> {
            try {
                task.run();
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}

輸出:

Threads size is 302

Thread name : Reference Handler

Thread name : Finalizer

Thread name : Signal Dispatcher

Thread name : main

Thread name : Monitor Ctrl-Break

Thread name : pool-1-thread-1

Thread name : pool-1-thread-2

Thread name : pool-1-thread-3

Thread name : pool-2-thread-1

Thread name : pool-2-thread-2

Thread name : pool-2-thread-3

Thread name : pool-3-thread-1

Thread name : pool-3-thread-2

Thread name : pool-3-thread-3

Thread name : pool-4-thread-1

Thread name : pool-4-thread-2

Thread name : pool-4-thread-3

Thread name : pool-5-thread-1

Thread name : pool-5-thread-2

Thread name : pool-5-thread-3

Thread name : pool-6-thread-1

Thread name : pool-6-thread-2

Thread name : pool-6-thread-3

…………

執行結果分析,執行緒數量302個,區域性執行緒池建立的核心執行緒沒有被回收。

修改初始化執行緒池部分:

//初始化一次執行緒池
threadPoolExecutor =
        new ThreadPoolExecutor(3, 15, 1000, TimeUnit.MINUTES, new LinkedBlockingDeque<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());

for (int i=1;i<100;i++){
    //使用執行緒池執行任務
    for(int j=0;j<10;j++){
        submit(new Runnable() {
            @Override
            public void run() {
            }
        });
    }

}

輸出:

Threads size is 8

Thread name : Reference Handler

Thread name : Finalizer

Thread name : Signal Dispatcher

Thread name : main

Thread name : Monitor Ctrl-Break

Thread name : pool-1-thread-1

Thread name : pool-1-thread-2

Thread name : pool-1-thread-3

解決方案

1、只初始化一次,每次執行worker複用執行緒池

2、每次執行完成後,關閉執行緒池

BackgroundWorker的定位是後臺執行worker均進行執行緒池的複用,所以採用方案1,每次在static靜態程式碼塊中初始化,使用時無需重新初始化。

解決後監控:

jvm記憶體監控,記憶體不再持續上升:

執行緒池恢復正常且平穩:

Jstack檔案,觀察執行緒池數量恢復正常:

Dump檔案分析執行緒池物件數量:

拓展

1、 如何關閉執行緒池

執行緒池提供了兩個關閉方法,shutdownNow 和 shutdown 方法。

shutdownNow方法的解釋是:執行緒池拒接收新提交的任務,同時立馬關閉執行緒池,執行緒池裡的任務不再執行。

shutdown方法的解釋是:執行緒池拒接收新提交的任務,同時等待執行緒池裡的任務執行完畢後關閉執行緒池。

2、 為什麼threadPoolExecutor不會被GC回收

threadPoolExecutor =
         new ThreadPoolExecutor(3, 15, 1000, TimeUnit.MINUTES, new LinkedBlockingDeque<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());

區域性使用後未手動關閉的執行緒池物件,會被GC回收嗎?獲取線上jump檔案進行分析:

發現執行緒池物件沒有被回收,為什麼不會被回收?檢視ThreadPoolExecutor.execute()方法:

如果當前執行緒數小於核心執行緒數,就會進入addWorker方法建立執行緒:

分析runWorker方法,如果存在任務則執行,否則呼叫getTask()獲取任務:

發現workQueue.take()會一直阻塞,等待佇列中的任務,因為Thread執行緒一直沒有結束, 存在引用關係:ThreadPoolExecutor->Worker->Thread,因為存在GC ROOT的引用,所以無法被回收 。