内存溢出问题排查(spring jpa缓存溢出)

1.问题和思路

      在最近的项目中遇到一个非常棘手的问题，系统在使用一段时间之后，就会直接down掉，经过排查error日志，发现了内存溢出的报错，报错如下:

      上边的报错的指的是内存频繁的GC垃圾回收但不能达到预期的回收效果，从而导致抛出内存溢出异常。
      刚开始以为是堆内存配置的比较小，查看了启动命令行参数，发现-Xms:256(堆内存最小的分配空间),-Xmx:512M(堆内存最大的分配空间)，这对于一个数据量在几十万的系统来说，确实设置的比较小，所以调整了-Xms:2048，-Xmx:2048。在生产环境中，最好设置-Xms和-Xmx参数为一致的，这样可以避免内存扩展而造成的系统功能延迟。 但是上述的命令经过一段时间之后，发现还是会出现内存溢出的问题。

      生产环境中一般会在启动项中添加-XX:HeapDumpOnOutOfMemoryError和-XX:HeapDumpPath,这两个参数在发生内存溢出时会打印当前的堆栈信息到-XX:HeapDumpPath指定的路径中，文件名格式一般为java_pid进程号.hprof，将该文件导出到本地种使用jprofiler进行解析和排查，分析的结果如下：

      从上图发现是因为使用hibernate的缓存执行计划太多而导致的内存溢出，已经占用了内存的92%，网上查询了下发现是因为配置了 spring.jpa.properties.hibernate.query.plan_cache_max_size和
spring.jpa.properties.hibernate.query.plan_parameter_metadata_max_size属性，配置如下：

查询资料需要将上述两个属性设置较小，默认为2048，我这里改为256，这次系统时间坚持的比较长了点，但是最终还是会出现内存溢出的问题。然后继续将堆内存快照的文件拿出来重新分析，还是出现BoundedConcurrentHashMap的内存占用率较高，总是这个导致内存溢出肯定有问题，要想彻底解决这个问题，就必须了解到BoundedConcurrentHashMap源码。

2.BoundedConcurrentHashMap

2.1. BoundedConcurrentHashMap特性

      BoundeConcurrentHashMap和Hastable类很相似，BoundedConcurrentHashMap提供了HashTable所有的功能规范，同事也针对HashTable做了补充。

1. HastTabel的get方法加了synchorizd同步方法的，但BoundedConcurrentHashMap的get方法没有Synchronized同步标志，所以有可能检索出的数据不适最新更新的数据。
2. BoundedConcurrentHashMap支持并发最大的线程数设置，通过将Map中的table[]数组按照concurrentLevel属性设置划分为多个段，多个段可以做到避免线程竞争。类似于java1.7的CourrentHashMap类。
3. BoundedConcurrentHashMap提供了缓存机制和淘汰策略，可以针对的实际场景选择相应的淘汰机制，正式由于该属性，Hibernate使用BoundedConcurrentHashMap作为缓存。

   2.2 BoundedConcurrentHashMap简单的源码分析

   2.2.1 BoundedConcurrentHashMap的初始化和属性值

         向学习其他的集合类型一样，分析源码的时候首先看它支持哪些属性和其初始化方法，方法如下：

   public class BoundedConcurrentHashMap<K, V> extends AbstractMap<K, V>
   		implements ConcurrentMap<K, V>, Serializable {
   private static final long serialVersionUID = 7249069246763182397L;
   
   //默认最大能容纳的元素数量
   static final int DEFAULT_MAXIMUM_CAPACITY = 512;
   //默认加载因子
   static final float DEFAULT_LOAD_FACTOR = 0.75f;
   //默认估计支持的并发线程数量
   static final int DEFAULT_CONCURRENCY_LEVEL = 16;
   //Map集合最大容纳元素数量
   static final int MAXIMUM_CAPACITY = 1 << 30;
   //预估最大支持的并发线程数量
   static final int MAX_SEGMENTS = 1 << 16; // slightly conservative
   
   static final int RETRIES_BEFORE_LOCK = 2;
   
   /* ---------------- Fields -------------- */
   //该字段主要作用是标志段的大小，用于快速的定位到插入的entry存在哪个段中
   final int segmentMask;
   //插入段中的偏移量
   final int segmentShift;
   //Map集合中的段数组，充当了分段锁的角色，继承了ReentranLock
   final Segment<K, V>[] segments;
   
   transient Set<K> keySet;
   transient Set<Map.Entry<K, V>> entrySet;
   transient Collection<V> values;
   
   /**
    * BoundedConcurrentHashMap的初始化方法
    * capacity：指定该集合现支持元素上限个数
    * concurrencyLevel：指定该集合预估支持的并发线程数，实际不一定是该值。
    * evictionStrategy: 指定集合中元素淘汰的算法
    * evictionListener: 指定监听器
    */
   public BoundedConcurrentHashMap(
           int capacity, int concurrencyLevel,
           Eviction evictionStrategy, EvictionListener<K, V> evictionListener) {
       if ( capacity < 0 || concurrencyLevel <= 0 ) {
           throw new IllegalArgumentException();
       }
       //设置预估支持并发线程的数量，选择capcaity/2和concurrencytLevel中最小的
       concurrencyLevel = Math.min( capacity / 2, concurrencyLevel ); // concurrencyLevel cannot be > capacity/2
       //设置的并发线程数量不能小于1
       concurrencyLevel = Math.max( concurrencyLevel, 1 );
       //判断容纳的元素上限个数是否满足
       if ( capacity < concurrencyLevel * 2 && capacity != 1 ) {
           throw new IllegalArgumentException( "Maximum capacity has to be at least twice the concurrencyLevel" );
       }
       //判断元素淘汰算法和监听器是否为空，若为空抛出异常
       if ( evictionStrategy == null || evictionListener == null ) {
           throw new IllegalArgumentException();
       }
       //因为预估支持的并发线程数依赖于段的角色，因此预估的并发线程数应小于默认最大的段大小
       if ( concurrencyLevel > MAX_SEGMENTS ) {
           concurrencyLevel = MAX_SEGMENTS;
       }
       //设置segment(段)大小和偏移量
       int sshift = 0;
       int ssize = 1;
       //满足segment(段)的大小小于设置的线程并发数，并且接近于2的次方的一个值。
       while ( ssize < concurrencyLevel ) {
           ++sshift;
           ssize <<= 1;
       }
       //计算段移位量，用于与hash进行移位运算，找到hash所在的段的位置
       segmentShift = 32 - sshift;
       segmentMask = ssize - 1;
       this.segments = Segment.newArray( ssize );
       //如果设置的容量大于最大的容量值，则将最大容量值赋予容量
       if ( capacity > MAXIMUM_CAPACITY ) {
           capacity = MAXIMUM_CAPACITY;
       }
       //设置每个段可以容纳的元素，满足小于capacity/sszie并且接近于2的次方。
       int c = capacity / ssize;
       int cap = 1;
       while ( cap < c ) {
           cap <<= 1;
       }
       //初始化每个段的大小
       for ( int i = 0; i < this.segments.length; ++i ) {
           this.segments[i] = new Segment<K, V>( cap, c, DEFAULT_LOAD_FACTOR, evictionStrategy, evictionListener );
       }
   }

         从上述方法可以看到BoundedConcurrentHashMap是一个支持固定元素数目，并且支持根据Eviction和EvictionListener来对过期的元素进行淘汰，并且支持分段锁，降低锁的竞争，可以完美的充当一个缓存的内部数据结构来。我们抱着学习的态度再继续深入下BoundedConcurrentHashMap的Segment创建，插入和获取方法。

   2.2.2 Semgent的初始化

static final class Segment<K, V> extends ReentrantLock {

    private static final long serialVersionUID = 2249069246763182397L;
    //当前段容纳的元素数量
    transient volatile int count;
    //修改的次数
    transient int modCount;
    //阈值
    transient int threshold;
    //元素数组
    transient volatile HashEntry<K, V>[] table;
    //加载因子
    final float loadFactor;
    //段的初始大小
    final int evictCap;
    //元素淘汰策略
    transient final EvictionPolicy<K, V> eviction;
    //监听器
    transient final EvictionListener<K, V> evictionListener;

    Segment(int cap, int evictCap, float lf, Eviction es, EvictionListener<K, V> listener) {
        loadFactor = lf;
        this.evictCap = evictCap;
        eviction = es.make( this, evictCap, lf );
        evictionListener = listener;
        setTable( HashEntry.<K, V>newArray( cap ) );
    }
    
    //设置段中元素数组和阈值
    void setTable(HashEntry<K, V>[] newTable) {
        threshold = (int) ( newTable.length * loadFactor );
        table = newTable;
    }

      上述的初始化方法没有什么复杂的地方，唯一可能存在疑问的是es.es.make( this, evictCap, lf )方法，该方法用于创建元素淘汰的策略，BoundedConcurrentHashMap实现了3种EvictionPolicy策略，如下：

• NullEvictionPolicy: 不做任何操作
• LRU：最近最少使用的淘汰
• LIRS：一种相对于LRU算法考虑到更全面的缓存算法，具体的可以查看缓存淘汰算法LIRS原理与实现

2.2.3 Segment的插入

//插入元素，若元素的值为空，则抛出异常
public V put(K key, V value) {
    if ( value == null ) {
        throw new NullPointerException();
    }
    int hash = hash( key.hashCode() );
    //获取该元素的hash值所对应的段，然后调用段的put方法，从而减少了锁的竞争。
    return segmentFor( hash ).put( key, hash, value, false );
}

/**
 * 段中插入元素
 * key: 元素key
 * hash: 元素key的hash值
 * value： 元素的value
 * onlyIfAbsent：是否需要替换已存在元素的值
 */
V put(K key, int hash, V value, boolean onlyIfAbsent) {
    lock();
    Set<HashEntry<K, V>> evicted = null;
    try {
        int c = count;
        //判断段中的元素是否已超过阈值并且未知道扩容策略，则调用rehash方法进行扩容。
        if ( c++ > threshold && eviction.strategy() == Eviction.NONE ) {
            rehash();
        }
        //遍历段中的元素信息，如果存在相同的元素，则赋值给e，否则返回空。
        HashEntry<K, V>[] tab = table;
        int index = hash & tab.length - 1;
        HashEntry<K, V> first = tab[index];
        HashEntry<K, V> e = first;
        while ( e != null && ( e.hash != hash || !key.equals( e.key ) ) ) {
            e = e.next;
        }
        //判断段中已是否存在该元素，若存在，则调用缓存算法策略重新计算缓存的元素内容。
        V oldValue;
        if ( e != null ) {
            oldValue = e.value;
            if ( !onlyIfAbsent ) {
                e.value = value;
                //计算缓存元素的内容
                eviction.onEntryHit( e );
            }
        }
        else {
            oldValue = null;
            ++modCount;
            //如果段内元素的数量大于初始化的上限并且淘汰策略不是Eviction.NONE时，执行eviction.execute()方法淘汰掉元素。
            count = c; // write-volatile
            if ( eviction.strategy() != Eviction.NONE ) {
                if ( c > evictCap ) {
                    // remove entries;lower count
                    evicted = eviction.execute();
                    // re-read first
                    first = tab[index];
                }
                //创建新的元素
                tab[index] = eviction.createNewEntry( key, hash, first, value );
                //重新计算新加入元素的淘汰机制
                Set<HashEntry<K, V>> newlyEvicted = eviction.onEntryMiss( tab[index] );
                if ( !newlyEvicted.isEmpty() ) {
                    if ( evicted != null ) {
                        evicted.addAll( newlyEvicted );
                    }
                    else {
                        evicted = newlyEvicted;
                    }
                }
            }
            else {
                tab[index] = eviction.createNewEntry( key, hash, first, value );
            }
        }
        return oldValue;
    }
    finally {
        unlock();
        notifyEvictionListener( evicted );
    }
}

      BoundedConcurrentHashMap的插入实际是Segment元素的插入，在Segment对象中插入元素时，首先需要获取到锁，这样对于一个BoundedConcurrentHashMap对象含有多个Segment对象可以有效的避免在并发环境中的锁竞争问题。
      Segment在插入元素的时候，需要考虑以下两种情况：

1. 若不存在淘汰机制，则在插入元素的时候需要判断Segment中的元素数量是否已超过初设的数量，若超过则进行扩容，扩容是将Segment中的table数组大小扩容一倍，然后重新分配之前已存在的元素到新的table数组中
2. 若存在淘汰机制，并且Segment中的元素数量已超过初设的数量，则调用淘汰算法机制的execute()方法，排查掉不满足条件的元素。然后再继续创建新的元素插入到Segment中，然后再重新计算段中元素是否满足淘汰机制。

      经过第二种情况设置淘汰算法策略，则会实现缓存所拥有的的特性，而像Hibernate中则就是设置了淘汰策略-LIRS, 从而实现的查询语句和参数的缓存。

2.2.4 Segment的获取

//从Map集合中通过key值获取结果
public V get(Object key) {
    int hash = hash( key.hashCode() );
    //获取到Segment，然后从其中获取到结果
    return segmentFor( hash ).get( key, hash );
}
/**
 * 从Segment中获取结果
 * key： Map集合中的键
 * hash: hash值
 */
V get(Object key, int hash) {
    int c = count;
    //判断段中的元素是否为0，若为0则直接返回null，标识Segment中不存在元素。
    if ( c != 0 ) { // read-volatile
        V result = null;
        //根据hash值获取定位到table中的位置。
        HashEntry<K, V> e = getFirst( hash );
        while ( e != null ) {
            if ( e.hash == hash && key.equals( e.key ) ) {
                V v = e.value;
                if ( v != null ) {
                    result = v;
                    break;
                }
                //如果Segment中不存在元素，则通过加锁重新获取下元素，查看是否能获取到元素，这部主要是弥补之前获取元素的时候，Map集合正在插入。
                else {
                    result = readValueUnderLock( e );
                    break;
                }
            }
            e = e.next;
        }
        //如果Segment存在该元素，则需要改变该元素在淘汰策略中的重要性，为之后的插入元素淘汰时做铺垫。
        if ( result != null ) {
            if ( eviction.onEntryHit( e ) ) {
                Set<HashEntry<K, V>> evicted = attemptEviction( false );
                notifyEvictionListener( evicted );
            }
        }
        return result;
    }
    return null;
}

      BoundedConcurrentHashMap元素的获取非常简单，只是在传统的元素获取上添加了淘汰策略机制，对命中的元素提高其重要性。

2.2.5 EvictionPolicy

      EvictionPolicy是BoundedConcurrentHashMap最重要的一个特性，用于淘汰集合中不重要的元素。BoundedConcurrentHashMap中实现了3个EvictionPolicy，包括LRU和LIRS,在Hibernate中使用LIRS淘汰算法。

      我们分析了造成内存溢出的实例BoundedConcurrentHashMap的实现，但发现如果设置了缓存的算法淘汰机制，则应该不会出现BoundedConcurrentHashMap的内存太大问题，我们再看看它的上级调用链-QueryPlanCache.

3.QueryPlanCache

      QueryPlanCache充当查询计划以及查询元数据的缓存。使用BoundedConcurrentHashMap作为底层实现，在QueryPlanCache的初始化方法总实例化了BoundedConcurrentHashMap。

public QueryPlanCache(final SessionFactoryImplementor factory) {
    this.factory = factory;
    //获取查询参数元数据支持元素的上限值(hibernate.query.plan_parameter_metadata_max_size)
    Integer maxParameterMetadataCount = ConfigurationHelper.getInteger(
            Environment.QUERY_PLAN_CACHE_PARAMETER_METADATA_MAX_SIZE,
            factory.getProperties()
    );
    if ( maxParameterMetadataCount == null ) {
        maxParameterMetadataCount = ConfigurationHelper.getInt(
                Environment.QUERY_PLAN_CACHE_MAX_STRONG_REFERENCES,
                factory.getProperties(),
                DEFAULT_PARAMETER_METADATA_MAX_COUNT
        );
    }
    //获取查询计划支持最大的上限(hibernate.query.plan_cache_max_size)
    Integer maxQueryPlanCount = ConfigurationHelper.getInteger(
            Environment.QUERY_PLAN_CACHE_MAX_SIZE,
            factory.getProperties()
    );
    if ( maxQueryPlanCount == null ) {
        maxQueryPlanCount = ConfigurationHelper.getInt(
                Environment.QUERY_PLAN_CACHE_MAX_SOFT_REFERENCES,
                factory.getProperties(),
                DEFAULT_QUERY_PLAN_MAX_COUNT
        );
    }

    queryPlanCache = new BoundedConcurrentHashMap( maxQueryPlanCount, 20, BoundedConcurrentHashMap.Eviction.LIRS );
    parameterMetadataCache = new BoundedConcurrentHashMap<>(
            maxParameterMetadataCount,
            20,
            BoundedConcurrentHashMap.Eviction.LIRS
    );

    nativeQueryInterpreter = factory.getServiceRegistry().getService( NativeQueryInterpreter.class );
}

      通过上述的QueryPlanCache方法，我们可以发现在application.properties中配置的spring.jpa.properties.hibernate.query.plan_parameter_metadata_max_size和spring.jpa.properties.hibernate.query.plan_cache_max_size属性在QueryPlanCache初始化中使用到了，但是和我们想的不一样的是这里的大小不是兆或者是字节大小，而是能够容纳元素的数量。
      综合以上，我们再看下之前内存溢出保存下来的快照和我们在application.propperties中配置的值，如下图：

      通过我们对堆内存的快照的分析，我们可以看到BoundenConcurrentHashMap元素数量达到73时已经沾满了内存，而我们在application.properties中配置的是256，所以达不到BoundedConcurrentHashMap的淘汰机制，新元素插入不进去，旧元素不能及时淘汰回收从而导致BoundedConcurrentHashMap越来越大直到内存溢出。
      通过将application.properties中的spring.jpa.properties.hibernate.query.plan_parameter_metadata_max_size和spring.jpa.properties.hibernate.query.plan_cache_max_size属性的值同时改为32，完美的解决了该问题。