背景

• 筆者的大數(shù)據(jù)監(jiān)控系統(tǒng)中有一項(xiàng)hdfs路徑下異常格式文件檢測(cè)的功能。簡(jiǎn)單的說就是每天需要定期的采集hdfs下的路徑涩禀。
• 在某天添加了hive staging路徑后亿汞，發(fā)現(xiàn)程序OOM了亮垫。當(dāng)時(shí)從代碼出發(fā)懷疑是(1)從DB查詢過大沒有分頁導(dǎo)致直接load到內(nèi)存導(dǎo)致OOM,(2)線程池中blockqueue是沒有設(shè)置大小的，可能任務(wù)都提交到blockqueue中導(dǎo)致內(nèi)存溢出躺涝。
• 基于上述兩個(gè)可能點(diǎn)對(duì)代碼進(jìn)行了修改并發(fā)布厨钻，但是在第二天還是又OOM了！诞挨！
• 于是只能對(duì)堆內(nèi)存進(jìn)行分析來找出真正的泄漏點(diǎn)莉撇。在這里僅分享最關(guān)鍵的一處泄漏點(diǎn)。
• 注： 可能有朋友問為何不通過解析fsimage來獲取hdfs詳情惶傻，其實(shí)fsimage的解析每天都有在做棍郎。筆者每天起docker容器拉取fsimage并解析后導(dǎo)入到hive分區(qū)表中，再進(jìn)行相關(guān)的加工后導(dǎo)入到mysql中银室，此過程雖已標(biāo)準(zhǔn)化但是比較麻煩涂佃。我也參考了hadoop回放fsimage代碼，通過java對(duì)fsimage進(jìn)行解析蜈敢，但是所需的JVM堆內(nèi)存需要很大(如：fsimage 20G則至少需要40G JVM內(nèi)存)辜荠。這也是沒有用java解析的原因，如果有你有更好的辦法麻煩跟我聯(lián)系抓狭，謝謝伯病。

排查過程

• 首先看下項(xiàng)目JVM參數(shù)。我使用的是G1回收期(確實(shí)給力)否过，然后有記錄相關(guān)的GC日志午笛，這個(gè)可以幫助我覺得到底要設(shè)置多大Xmx和Xms，然后在程序OOM的時(shí)候會(huì)自動(dòng)給我dump下堆內(nèi)存(雖然dump過程中對(duì)程序有影響苗桂，但是好像沒其他更好辦法了Ｒ┗恰！)

${JAVA_EXEC} -server -XX:+UseG1GC -Xmx8G -Xms8G -Xss256k -XX:+HeapDumpOnOutOfMemoryError  -XX:HeapDumpPath=/var/log/xxx  -XX:MaxGCPauseMillis=300 -Xloggc:/var/log/xxx/xxx_gc.log  -XX:+PrintGCTimeStamps -XX:+PrintGCDetails   -Dservice.name=${EXEC_COMMEN} -Dfastjson.parser.safeMode=true   -cp "${FULL_PATH_SERVER_JAR}:${LIB_PATH}/*:${CONFIG_VERSION_PATH}/" ${MAIN_CLASS} >> ${LOG_FILE} 2>&1 &

• 在dump程序之前煤伟，我先用jmap -histo:live <PID>執(zhí)行了一次強(qiáng)制的full GC癌佩，之后通過arthas的heapdump 命令dump下來，也可以用jmap的dump命令便锨。

• 關(guān)于OOM分析工具围辙，visualvm和eclipse memory analyze我都有用，總體上覺得MemoryAnalyze會(huì)好用一些放案。

• 如下圖所示是程序中加載類的清單姚建，可以看到hdfs的configuration占據(jù)了絕大多數(shù)

  
  
  image.png

• 同時(shí)看到重點(diǎn)包加載也都是跟hadoop的filesystem相關(guān)，因此可以判斷這個(gè)可能是主要的泄漏點(diǎn)卿叽。
  

  image.png

• 其它的還可以看到很多kerberos鑒權(quán)類也沒有釋放

  
  
  image.png

代碼排查和修改

• 在代碼的工具類中有個(gè)獲取filesystem的方法桥胞。代碼通過cluster對(duì)象來攜帶集群的詳情，然后構(gòu)建FileSystem的過程(很多線程都會(huì)調(diào)用此方法)考婴，每個(gè)進(jìn)程進(jìn)來都先初始化一個(gè)Configuration對(duì)象贩虾，然后進(jìn)行kerberos鑒權(quán)！相關(guān)代碼如下：

 public FileSystem getFileSystemInstance(Cluster cluster){
        FileSystem fs = null;
        HadoopClusterParam param = JSONObject.parseObject(cluster.getParam(), HadoopClusterParam.class);
        Configuration conf = new Configuration();
        System.setProperty("java.security.krb5.conf", param.getKrb5Conf());
        conf.set("dfs.namenode.kerberos.principal", param.getHdfsKerberosPrincipal());
        conf.set("dfs.namenode.kerberos.principal.pattern", "*");
        conf.set("hadoop.security.authentication", "kerberos");
        conf.set("fs.trash.interval", "1");
        conf.set("fs.defaultFS", String.format("hdfs://%s", param.getHaName()));
        conf.set(String.format("dfs.client.failover.proxy.provider.%s", param.getHaName()), "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
        conf.set(String.format("dfs.ha.namenodes.%s", param.getHaName()), param.getHaNamenodes());
        String[] nns = param.getHaNamenodes().split(",");
        String[] nnHosts = param.getNamenodeAddress().split(",");
        conf.set(String.format("dfs.namenode.rpc-address.%s.%s", param.getHaName(), nns[0]), String.format("%s:8020", nnHosts[0]));
        conf.set(String.format("dfs.namenode.rpc-address.%s.%s", param.getHaName(), nns[1]), String.format("%s:8020", nnHosts[1]));
        conf.set("dfs.nameservices", param.getHaNames());
        try {
            UserGroupInformation.setConfiguration(conf);
            UserGroupInformation.loginUserFromKeytab(param.getHdfsKerberosPrincipal(), param.getHdfsKerberosKeytab());
            fs =  FileSystem.get(conf);
        } catch (Exception e) {
            LOGGER.error("Build FileSystem found execption,caused by:", e);
        }
        return fs;
    }

• 上面代碼有兩個(gè)致命缺點(diǎn)(1)Configuration其實(shí)是可以復(fù)用的而不需要每次重新new沥阱，(2)等于登陸一次kerberos鑒權(quán)即可而不需要每個(gè)都用UserGroupInformation去鑒權(quán)缎罢。
• 基于上面的致命缺點(diǎn)進(jìn)行修改后的代碼如下：

    private Map<String,Configuration> confMap = new HashMap<>();

    private Configuration generateFileSystemConf(Cluster cluster) throws IOException {

        UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();
        String userName = currentUser.getUserName();
        //防止其它操作更新掉當(dāng)前線程中的kerberos認(rèn)證用戶
        if(!HDFS_USER.equals(userName)){
            LOGGER.info("The login user has changed,current user:{},change to {}",userName,HDFS_USER);
            confMap.remove(cluster.getClusterName());
        }
        if(confMap.getOrDefault(cluster.getClusterName(),null)==null){
            Configuration conf = new Configuration();
            HadoopClusterParam param = JSONObject.parseObject(cluster.getParam(), HadoopClusterParam.class);
            System.setProperty("java.security.krb5.conf", param.getKrb5Conf());
            conf.set("dfs.namenode.kerberos.principal", param.getHdfsKerberosPrincipal());
            conf.set("dfs.namenode.kerberos.principal.pattern", "*");
            conf.set("hadoop.security.authentication", "kerberos");
            conf.set("fs.trash.interval", "1");
            conf.set("fs.defaultFS", String.format("hdfs://%s", param.getHaName()));
            conf.set(String.format("dfs.client.failover.proxy.provider.%s", param.getHaName()), "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
            conf.set(String.format("dfs.ha.namenodes.%s", param.getHaName()), param.getHaNamenodes());
            String[] nns = param.getHaNamenodes().split(",");
            String[] nnHosts = param.getNamenodeAddress().split(",");
            conf.set(String.format("dfs.namenode.rpc-address.%s.%s", param.getHaName(), nns[0]), String.format("%s:8020", nnHosts[0]));
            conf.set(String.format("dfs.namenode.rpc-address.%s.%s", param.getHaName(), nns[1]), String.format("%s:8020", nnHosts[1]));
            conf.set("dfs.nameservices", param.getHaNames());
            UserGroupInformation.setConfiguration(conf);
            UserGroupInformation.loginUserFromKeytab(param.getHdfsKerberosPrincipal(), param.getHdfsKerberosKeytab());
            confMap.put(cluster.getClusterName(),conf);
            return conf;
        }
        return confMap.get(cluster.getClusterName());
    }

    public FileSystem getFileSystemInstance(Cluster cluster){
        FileSystem fs = null;
        try {
            Configuration conf = this.generateFileSystemConf(cluster);
            fs =  FileSystem.get(conf);
        } catch (Exception e) {
            LOGGER.error("Build FileSystem found execption,caused by:", e);
        }
        return fs;
    }

• 修改后重新發(fā)布程序，運(yùn)行了幾天之后再也沒有OOM的現(xiàn)象考杉。從下圖可見每次minor gc都能正常的回收策精，old_gen也維持在一個(gè)較低的范圍內(nèi)。

  
  
  image.png

后記

• 為什么Configuration和UserGroupInformation無法回收的問題崇棠，我猜測(cè)可能跟FileSystem的設(shè)計(jì)有關(guān)咽袜。
• Hadoop把對(duì)于文件系統(tǒng)的調(diào)用封裝成了一個(gè)FileSystem類，同時(shí)FileSystem對(duì)于文件系統(tǒng)類的實(shí)例做了緩存枕稀，如果是來自同一個(gè)文件系統(tǒng)询刹，它會(huì)返回同一個(gè)實(shí)例。代碼如下：

  public static FileSystem get(URI uri, Configuration conf) throws IOException {
    String scheme = uri.getScheme();
    String authority = uri.getAuthority();

    if (scheme == null && authority == null) {     // use default FS
      return get(conf);
    }

    if (scheme != null && authority == null) {     // no authority
      URI defaultUri = getDefaultUri(conf);
      if (scheme.equals(defaultUri.getScheme())    // if scheme matches default
          && defaultUri.getAuthority() != null) {  // & default has authority
        return get(defaultUri, conf);              // return default
      }
    }
    
    String disableCacheName = String.format("fs.%s.impl.disable.cache", scheme);
    if (conf.getBoolean(disableCacheName, false)) {
      return createFileSystem(uri, conf);
    }

    return CACHE.get(uri, conf);
  }

• 為什么要設(shè)置cache萎坷？我猜是因?yàn)槊總€(gè)FileSystem的實(shí)例都會(huì)建立一個(gè)到HDFS Namenode的連接凹联，在大量并發(fā)的讀時(shí)緩存確實(shí)能減低namenode的壓力。那么可能產(chǎn)生什么情況哆档？你從FileSystem里面拿到的文件系統(tǒng)的實(shí)例可能別人也拿到了蔽挠，而且可能正在用，所以這里也是不推薦你close掉fs的原因
• 由于FileSystem是緩存在hadoop中的瓜浸，則對(duì)Configuration和UserGroupInformation的引用關(guān)系是存在的澳淑，也即GC ROOT是存在的，因此當(dāng)前不會(huì)被回收斟叼。慢慢堆積也就OOM了偶惠。