Ribbon的ServerStats引起内存泄露问题总结

问题描述

服务运行一段时间之后，出现页面卡顿加载慢的问题，使用top命令查看了服务器的使用情况，发现CPU飙高，接着查看了该进程中每个线程的占用情况，发现导致CPU高的线程是JVM垃圾回收的线程，然后使用jstat命令打印了GC的情况，基本隔几秒就进行一次FULL GC，每次FULL GC之后有大量的内存空间释放不掉，所以JVM内存空间很快又被耗尽再次进行GC。

既然JVM在频繁的进行垃圾回收，接下来就要分析是什么原因造成的，使用jmap命令导出了一份内存快照，导入到Eclipse Memery Analyzer中进行分析，可以看到已经有潜在的内存泄露问题了：

从Problem Suspect2中可以看出引起内存泄露的原因很可能是com.netxflix.stats.distribution的DataDistribution有关，DataDistribution中创建了一个定时任务线程池ScheduledThreadPoolExecutor，在周期性的执行任务。

接下来再看下查看下占用率最大的对象：

占用内存空间最多的是一些double数组，查看一下它的引用链，果然是和com.netxflix.stats包下的类有关：

com.netxflix.stats下的类在netflix-statistics包中，netflix-statistics是Ribbon对Server各个维度进行数据统计的一个模块，比如服务的响应时间之类的，在负载均衡的时候就可以根据收集到的Server统计信息可以判断出选择哪个Server合适。

然而项目中并没有直接使用Ribbon，然后查看了maven的引用关系，发现是公司自研的一个上传/下载文件的组件，里面使用了Ribbon进行负载均衡，选择合适的服务进行上传/下载，接下来就去看看代码，是不是代码有什么BUG或者我们的使用方式不对造成了内存泄露。

由于是自研的组件，代码未开源，接下来就以简化的方式描述大概的处理逻辑。

在上传/下载文件的方法中，第一步是获取MyLoadBalancerContext ，MyLoadbalacerContext是对ribbon组件LoadbalacerContext的一个封装：

public class MyLoadbalacerContext {

    // 省略了其他属性

    // netflix的LoadBalancerContext对象

    LoadBalancerContext loadBalancerContext;

    // 用来验证MD5使用

    String md5;

}

MyLoadbalacerContext获取逻辑大概如下（对代码进行了简化）：

参数中传入一个key、storageId存储ID和right权限关键字，key用来从缓存中获取MyLoadbalacerContext对象，storageId和right用来对比MD5与缓存中取到的MyLoadbalacerContext是否一致
如果缓存中包含key并且md5一致返回缓存中的MyLoadbalacerContext对象，如果缓存中存在但是MD5不一致，清除缓存中的对象，并通过MyLoadbalacerContext中的LoadBalancerContext获取LoadBalancer对象，调用shutdown方法关闭一些资源
如果缓存中不存在或者MD5不一致，接下来都会新创建一个MyLoadbalacerContext

   // LoadBalancerContext缓存

    private Map<String, MyLoadbalacerContext> cacheMap;

    /**

     *

     * @param key 缓存Key

     * @param storageId 存储ID

     * @param right 读写权限关键字

     * @return

     */

    public MyLoadbalacerContext loadMyLoadbalacerContext(String key, String storageId, String right) {

        if (cacheMap.containsKey(key)) {

           MyLoadbalacerContext context = cacheMap.get(key);

           // 计算md5

           String targetMd5 = calculateMd5(storageId, right);

           if (context.getMd5().equals(targetMd5)) {

               return context;

           }

           // 缓存中移除

           cacheMap.remove(key);

           // 获取BaseLoadBalancer

           BaseLoadBalancer baseLoadBalancer = (BaseLoadBalancer)context.getLoadBalancerContext().getLoadBalancer();

           // 关闭一些资源

           baseLoadBalancer.shutdown();

        }

        synchronized (this) {

            // 新建MyLoadbalacerContext

            MyLoadbalacerContext context = createMyLoadbalacerContext(otherInfo);

            cacheMap.put(key, context);

            return context;

        }

    }

    public MyLoadbalacerContext createMyLoadbalacerContext(String storageId, String right) {

        MyLoadbalacerContext context = new MyLoadbalacerContext();

        // 计算MD5, 忽略细节，注意这里和calculateMd5方法中的区别，少了filterByStorageId步骤

        String serverId = getAllServerConfig().filterByRight(right);

        context.setMd5(calculateMd5(right));

        return context;

    }

    /**

     * 根据StorageId和权限计算MD5

     * @param storageId

     * @param right 权限

     * @return

     */

    public String calculateMd5(String storageId, String right) {

        // 根据storageId和right获取serverID

        String serverID = getAllServerConfig().filterByStorageId(storageId).filterByRight(right);

        return calculateMd5(serverId);

    }

    public String calculateMd5(String content) {

        // 省略了计算MD5的方法

    }

由于代码做了简化，可能一眼就发现了问题，在创建createMyLoadbalacerContext的方法中，只通过了权限获取serverID来计算MD5，而与缓存中的context进行对比时，是通过storageId和权限来获取serverId做的MD5计算，假如配置文件中配置了多个存储服务，两个MD5很可能就不一致了。

当然实际的代码要复杂的多，所以一开始看代码并没有发现问题的所在，在调式的过程中发现每次从缓存中对比MD5都不一致，仔细研究了代码，才发现创建MyLoadbalacerContext时设置MD5的方式存在问题。

既然已经发现了代码的BUG，那么问题已经逐渐清晰了，每次与缓存中的对象对比MD5时都不一致，所以每次都会新生成一个MyLoadbalacerContext，在大量用户进行上传/下载文件的时候，频繁的创建对象，但是还有一个疑问未解决，对象已经从缓存中清除了，并且调用了shutdown方法关闭了资源，为什么还有大量的Server统计信息回收不掉呢，想要解开这个疑问，只能去研究一下Ribbon的源码了。

RibbonLoadBalancerClient

为了节省篇幅，从RibbonLoadBalancerClient开始看起，如何执行到这里的过程可以参考之前写的文章：【Spring Cloud】Ribbon调用过程

在RibbonLoadBalancerClient的execute方法中:

获取ILoadBalancer，ILoadBalancer是负载均衡规则的父类
通过ILoadBalancer的chooseServer方法选择服务
调用第二个execute方法执行diam

在第二个execute方法中，创建了RibbonStatsRecorder对象，并调用recordStats方法记录统计信息，以便在负载均衡的时候根据每个Server的负载情况选出合适的Server：

public class RibbonLoadBalancerClient implements LoadBalancerClient {

  // 首先进入第一个execute方法

    public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint)

            throws IOException {

        // 获取LoadBalancer实现类

        ILoadBalancer loadBalancer = getLoadBalancer(serviceId);

        // 根据负载均衡规则选取一个服务

        Server server = getServer(loadBalancer, hint);

        if (server == null) {

            throw new IllegalStateException("No instances available for " + serviceId);

        }

        RibbonServer ribbonServer = new RibbonServer(serviceId, server,

                isSecure(server, serviceId),

                serverIntrospector(serviceId).getMetadata(server));

        // 调用execute方法执行请求

        return execute(serviceId, ribbonServer, request);

    }

    protected Server getServer(ILoadBalancer loadBalancer, Object hint) {

        if (loadBalancer == null) {

            return null;

        }

        // 调用chooseServer选择一个服务

        return loadBalancer.chooseServer(hint != null ? hint : "default");

    }

  // 进入第二个execute方法

  @Override

	public <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException {

		Server server = null;

		if(serviceInstance instanceof RibbonServer) {

			server = ((RibbonServer)serviceInstance).getServer();

		}

		if (server == null) {

			throw new IllegalStateException("No instances available for " + serviceId);

		}

		RibbonLoadBalancerContext context = this.clientFactory

				.getLoadBalancerContext(serviceId);

		// 创建RibbonStatsRecorder，用于记录统计信息

		RibbonStatsRecorder statsRecorder = new RibbonStatsRecorder(context, server);

		try {

			T returnVal = request.apply(serviceInstance);

			// 记录统计信息

			statsRecorder.recordStats(returnVal);

			return returnVal;

		}

		// ...

		return null;

	}

}

LoadBalancerContext

LoadBalancerContext中引用了ILoadBalancer，在创建LoadBalancerContext时需要指定ILoadBalancer具体的实现类：

public class LoadBalancerContext implements IClientConfigAware {

    private ILoadBalancer lb;

    public LoadBalancerContext(ILoadBalancer lb, IClientConfig clientConfig, RetryHandler handler) {

        this(lb, clientConfig);

        this.defaultRetryHandler = handler;

    }

}

RibbonStatsRecorder

RibbonStatsRecorder中引用了Ribbon负载均衡上下文对象RibbonLoadBalancerContext，在构造函数中，通过RibbonLoadBalancerContext获取了ServerStats对象：

public class RibbonStatsRecorder {

  // 负载均衡Context

	private RibbonLoadBalancerContext context;

  // 服务统计信息

	private ServerStats serverStats;

	private Stopwatch tracer;

	public RibbonStatsRecorder(RibbonLoadBalancerContext context, Server server) {

		this.context = context;

		if (server != null) {

      // 获取当前服务的统计信息

			serverStats = context.getServerStats(server);

			context.noteOpenConnection(serverStats);

			tracer = context.getExecuteTracer().start();

		}

	}

	public void recordStats(Object entity) {

		this.recordStats(entity, null);

	}

	public void recordStats(Throwable t) {

		this.recordStats(null, t);

	}

	protected void recordStats(Object entity, Throwable exception) {

		if (this.tracer != null && this.serverStats != null) {

			this.tracer.stop();

			long duration = this.tracer.getDuration(TimeUnit.MILLISECONDS);

			this.context.noteRequestCompletion(serverStats, entity, exception, duration, null/* errorHandler */);

		}

	}

}

看到ServerStats是不是很眼熟，在分析内存快照的时候，那些double变量就是被ServerStats所引用的，接下来就看一下RibbonLoadBalancerContext中是如何实现getServerStats方法的。

LoadBalancerContext

getServerStats方法在LoadBalancerContext中，获取步骤如下：

获取ILoadBalancer
将ILoadBalancer转为AbstractLoadBalancer，调用getLoadBalancerStats获取LoadBalancerStats对象
通过LoadBalancerStats的getSingleServerStat获取ServerStats

public class LoadBalancerContext implements IClientConfigAware {

    public final ServerStats getServerStats(Server server) {

        ServerStats serverStats = null;

        ILoadBalancer lb = this.getLoadBalancer();

        if (lb instanceof AbstractLoadBalancer){

            // 从LoadBalancer中获取统计信息

            LoadBalancerStats lbStats = ((AbstractLoadBalancer) lb).getLoadBalancerStats();

            serverStats = lbStats.getSingleServerStat(server);

        }

        return serverStats;

    }

}

LoadBalancerStats

进入到LoadBalancerStats的getSingleServerStat方法，可以看到它使用了Guava的LoadingCache对数据进行缓存，getSingleServerStat会从缓存中获取对应的ServerStats统计对象：

如果缓存中存在该Server的统计对象，直接返回
如果不存在，将会调用createServerStats生成一个ServerStats对象并放入缓存

public class LoadBalancerStats {

    // 存放ServerStats的缓存

    private final LoadingCache<Server, ServerStats> serverStatsCache =

        CacheBuilder.newBuilder()

            .expireAfterAccess(SERVERSTATS_EXPIRE_MINUTES.get(), TimeUnit.MINUTES)// 缓存失效时间

            .removalListener(new RemovalListener<Server, ServerStats>() {

                @Override

                public void onRemoval(RemovalNotification<Server, ServerStats> notification) {

                    notification.getValue().close();

                }

            })

            .build(

                new CacheLoader<Server, ServerStats>() {

                    public ServerStats load(Server server) {

                        // 调用createServerStats生成ServerStats对象

                        return createServerStats(server);

                    }

                });

   public ServerStats getSingleServerStat(Server server) {

        return getServerStats(server);

   }

   private ServerStats getServerStats(Server server) {

        try {

            // 从缓存中获取Server的ServerStats对象，如果为空将会调用createServerStats生成一个ServerStats对象

            return serverStatsCache.get(server);

        } catch (ExecutionException e) {

            ServerStats stats = createServerStats(server);

            serverStatsCache.asMap().putIfAbsent(server, stats);

            return serverStatsCache.asMap().get(server);

        }

    }

}

ServerStats

接下来看看ServerStats在初始化的时候都做了什么：

创建了DataDistribution对象dataDist
创建了DataPublisher对象，之后调用了它的start的方法并传入第一步中创建的dataDist对象，DataPublisher听名字像和数据发布有关的，那么start方法应该是启动了什么东西

public class ServerStats {

    private DataDistribution dataDist = new DataDistribution(1, PERCENTS); // in case

    private DataPublisher publisher = null;

    private final Distribution responseTimeDist = new Distribution();

   /**

     * Initializes the object, starting data collection and reporting.

     */

    public void initialize(Server server) {

        serverFailureCounts = new MeasuredRate(failureCountSlidingWindowInterval);

        requestCountInWindow = new MeasuredRate(300000L);

        if (publisher == null) {

            dataDist = new DataDistribution(getBufferSize(), PERCENTS);

            // 创建DataPublisher

            publisher = new DataPublisher(dataDist, getPublishIntervalMillis());

            publisher.start();

        }

        this.server = server;

    }

}

DataPublisher

进入到DataPublisher的start方法，它启动了一个定时执行任务的线程池，并创建了需要执行的任务，在任务中调用了DataPublisher的publish方法进行数据统计：

public class DataPublisher {

    private static final String THREAD_NAME = "DataPublisher";

    private static final boolean DAEMON_THREADS = true;

    private static ScheduledExecutorService sharedExecutor = null;

    private final DataAccumulator accumulator;

    private final long delayMillis;

    private Future<?> future = null;

    public DataPublisher(DataAccumulator accumulator, long delayMillis) {

        this.accumulator = accumulator;

        this.delayMillis = delayMillis;

    }

    public DataAccumulator getDataAccumulator() {

        return this.accumulator;

    }

    public synchronized boolean isRunning() {

        return this.future != null;

    }

    public synchronized void start() {

        if(this.future == null) {

            // 创建任务

            Runnable task = new Runnable() {

                public void run() {

                    try {

                        // 调用DataAccumulator的publish发布任务

                        DataPublisher.this.accumulator.publish();

                    } catch (Exception var2) {

                        DataPublisher.this.handleException(var2);

                    }

                }

            };

            // 初始化定时任务线程池

            this.future = this.getExecutor().scheduleWithFixedDelay(task, this.delayMillis, this.delayMillis, TimeUnit.MILLISECONDS);

        }

    }

}

DataAccumulator

DataAccumulator中引用了两个DataBuffer对象，分别为current和previous，在publish方法中调用了current的startCollection开始数据收集：

public abstract class DataAccumulator implements DataCollector {

    private DataBuffer current;

    private DataBuffer previous;

    private final Object swapLock = new Object();

    @SuppressWarnings({"MDM_WAIT_WITHOUT_TIMEOUT"})

    public void publish() {

        DataBuffer tmp = null;

        Lock l = null;

        Object var3 = this.swapLock;

        synchronized(this.swapLock) {

            tmp = this.current;

            this.current = this.previous;

            this.previous = tmp;

            l = this.current.getLock();

            l.lock();

            try {

                // 开始执行统计

                this.current.startCollection();

            } finally {

                l.unlock();

            }

            l = tmp.getLock();

            l.lock();

        }

        try {

            tmp.endCollection();

            this.publish(tmp);

        } finally {

            l.unlock();

        }

    }

}

DataBuffer

在DataBuffer中看到了我们想要找的double数组：

public class DataBuffer extends Distribution {

    private final Lock lock = new ReentrantLock();

    // double数组

    private final double[] buf;

    private long startMillis;

    private long endMillis;

    private int size;

    private int insertPos;

    public DataBuffer(int capacity) {

        this.buf = new double[capacity];

        this.startMillis = 0L;

        this.size = 0;

        this.insertPos = 0;

    }

    public Lock getLock() {

        return this.lock;

    }

    public int getCapacity() {

        return this.buf.length;

    }

    public long getSampleIntervalMillis() {

        return this.endMillis - this.startMillis;

    }

    public int getSampleSize() {

        return this.size;

    }

    public void clear() {

        super.clear();

        this.startMillis = 0L;

        this.size = 0;

        this.insertPos = 0;

    }

    public void startCollection() {

        this.clear();

        this.startMillis = System.currentTimeMillis();

    }

}

总结

先不管具体的统计实现细节，猜测一下内存对象无法回收的原因：

由于每次上传/下载都新生成了MyLoadBalancerContext，MyLoadBalancerContext中引用了com.netflix.loadbalancer下的LoadBalancerContext对象，相当于每次也生成了新的LoadBalancerContext对象；
LoadBalancerContext中引用了ILoadBalancer，获取Server是通过ILoadBalancer的chooseServer方法实现的，由于LoadBalancerContext每次都生成了新的对象，ILoadBalancer在chooseServer时也生成了不同的Server对象，尽管Server对象中的ip port等信息是一致的；
由于Server对象也是每次都重新生成，导致在LoadingCache缓存中无法获取上次缓存的数据，ServerStats也跟着重新生成；
大量的ServerStats对象不断在重新生成，ServerStats中引用了DataPublisher，DataPublisher中又使用了线程池定时执行任务，尽管缓存中设置的有失效时间，由于线程池未关闭，一直处于运行状态，所以即便缓存失效对象也并不能被垃圾回收器所回收；
虽然业务代码中调用了BaseLoadBalancer的shutdown方法进行资源关闭，但是并未关闭DataPublisher的线程池；
以上的连锁反应，最终导致内存空间耗尽，频繁进行GC；

BaseLoadBalancer

public class BaseLoadBalancer extends AbstractLoadBalancer implements

        PrimeConnections.PrimeConnectionListener, IClientConfigAware {

    public void shutdown() {

        cancelPingTask();

        if (primeConnections != null) {

            primeConnections.shutdown();

        }

        Monitors.unregisterObject("LoadBalancer_" + name, this);

        Monitors.unregisterObject("Rule_" + name, this.getRule());

    }

}

解决方案

修改代码，在创建MyLoadBalancerContext时设置MD5的方式和从缓存中获取时保持一致，这样不仅可以利用缓存，也不会因为每次都生成新的对象导致内存泄露：

    public MyLoadbalacerContext createMyLoadbalacerContext(String storageId, String right) {

        MyLoadbalacerContext context = new MyLoadbalacerContext();

        // 计算MD5,注意与缓存获取时计算MD5的方式保持一致

        String serverId = getAllServerConfig().filterByStorageId(storageId).filterByRight(right);

        context.setMd5(calculateMd5(right));

        return context;

    }

巴特西