线程池引起的jvm内存过大问题

     之前的一个hbase表结构和rowkey规划不合理，我重新设计了一个新的hbase表，需要把旧表的数据写入到新表中，采用的方案是一个region一个region的倒数据，这样旧表的读是scan顺序读，新表的写是随机写，整体速度相对较快。

    读采用单线程，写采用线程池（Executors.newFixedThreadPool()），改进scan查询速度的caching配置设置为500，写线程池设置为100，在循环scan结果集时，每条数据的写新表的操作均会作为一个线程任务提交给线程池，由于没有充分的评估scan（产生数据）和写（消费数据）的速度，导致查询数据的速度比写的速度快，每次scan查询的数据均不能完全处理掉，因此导致线程池的队列不断的累积任务，在半小时内就导致消耗jvm内存到达16G(每个写操作均会持有scan出的数据，因此比较耗内存)，发现问题后修改线程池大小为300，caching设置为200，并在每次scan查询时休眠100ms，这时基本上每次查询出的数据均会完全写入hbase，平均处理数据的速度由全速的10万条/s，降为8.8万/s，速度略有下降，但是不会因为任务堆积而导致jvm暴涨，最终出现OOM。

    一般的业务场景可能提交任务到线程池的量或者任务占用的内存不会太大，而且线程池与队列并没有字面上的直接联系，因此程序猿很容易忽略队列最常见的生产和消费问题，导致程序在运行一段时间后就会出现莫名的OOM。

  

   关于线程池ThreadPoolExecutor中队列的JDK解释：

   

   所有 BlockingQueue 都可用于传输和保持提交的任务。可以使用此队列与池大小进行交互：

• 如果运行的线程少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。
• 如果运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列，而不添加新的线程。
• 如果无法将请求加入队列，则创建新的线程，除非创建此线程超出 maximumPoolSize，在这种情况下，任务将被拒绝。

排队有三种通用策略：

1. 直接提交。工作队列的默认选项是 SynchronousQueue，它将任务直接提交给线程而不保持它们。在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
2. 无界队列。使用无界队列（例如，不具有预定义容量的 LinkedBlockingQueue）将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。这样，创建的线程就不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
3. 有界队列。当使用有限的 maximumPoolSizes 时，有界队列（如 ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是 I/O 边界），则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，CPU 使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。

    通过Executors的newFixedThreadPool()方法实现如下：

   public static ExecutorService newFixedThreadPool(int nThreads) {

        return new ThreadPoolExecutor(nThreads, nThreads,

                                      0L, TimeUnit.MILLISECONDS,

                                      new LinkedBlockingQueue<Runnable>());

    }

   可见采用的是无界队列，在使用该方法创建的线程池时需要注意队列的堆积问题，我的解决办法是：

 

 //MAX_THREAD_SIZE：线程池大小

  if ( pool.getQueue().size()  > MAX_THREAD_SIZE*3 ) {

     LOG.error("暂停提交任务到线程池，堆积：" +pool.getQueue().size());

     while(true) {

         if ( pool.getQueue().size()  < MAX_THREAD_SIZE ) {

             break;

         }

     Thread.sleep(5);

   }

   LOG.error("终止暂停，堆积：" +pool.getQueue().size());

 }

 pool.execute(new PutTask(puts, lastRow));

 

对JDK的ThreadPoolExecutor类的源码解读如下（1.6版的源码，1.7的版本实现略有不同）：

 

 

//执行任务，如果线程数>=核心池大小则把任务放入队列，如果放入失败则查询线程数是否已经>=max池大小，如果没有则创建新的线程，如果线程数<核心池大小，则直接创建新的线程，线程启动后执行完一个任务可能并不会直接消亡，会有若干场景，具体逻辑请参看后面的run()，getTask()， workerCanExit()三个方法

public void execute(Runnable command) {

        if (command == null)

            throw new NullPointerException();

        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {

//如果workQueue是SynchronousQueue，则如果没有线程在等待从queue中获取任务，则会放置失败，返回false，因为SynchronousQueue队列是不缓存数据的，因此对于SynchronousQueue队列的线程池，如果maxPoolSize很大，则基本上有多少个任务就会产生多少个线程，好处是可以立即处理业务，坏处是cpu和jvm内存均会有较大的消耗，创建过多的线程极有可能导致OOM

            if (runState == RUNNING && workQueue.offer(command)) {

                if (runState != RUNNING || poolSize == 0)

                    ensureQueuedTaskHandled(command);

            }

            else if (!addIfUnderMaximumPoolSize(command))

                reject(command); // is shutdown or saturated

        }

}

 

//执行完一个任务后，线程会尝试从任务队列中获取新的任务，如果获取不到则该线程终止，可见ThreadPoolExecutor线程池维护一定量的线程存活的方法是让线程一直有事可做，可参看后续的getTask()方法

public void run() {

            try {

                Runnable task = firstTask;

                firstTask = null;

                while (task != null || (task = getTask()) != null) {

                    runTask(task);

                    task = null;

                }

            } finally {

                workerDone(this);

            }

        }

 

//这是一个可永远进行的循环，退出条件如下（需结合后续的workerCanExit()方法）：

//1、线程池已经STOP或者TERMINATED，该线程消亡；

//2、线程池处于SHUTDOWN状态，如果任务队列有任务则返回任务，线程继续存活；

//3、如果线程数>coreSize或者allowCoreThreadTimeOut==true，这时如果在超时前可以从队列中获取到任务则返回，线程继续存活；

//4、如果不满足以上3个条件，则调用队列的take()阻塞方法，直到有队列中有任务，该处可以确保线程池中一直有<=corePoolSize的线程存在，当然如果allowCoreThreadTimeOut==true则不会采用队列的阻塞方法；

//5、执行完以上4个逻辑后，如果线程没有获取到任务，则会调用workerCanExit()方法检查该线程是否符合退出条件：请参看workerCanExit()方法

Runnable getTask() {

        for (;;) {

            try {

                int state = runState;

                if (state > SHUTDOWN)

                    return null;

                Runnable r;

                if (state == SHUTDOWN)  // Help drain queue

                    r = workQueue.poll();

                else if (poolSize > corePoolSize || allowCoreThreadTimeOut)

                    r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);

                else

                    r = workQueue.take();

                if (r != null)

                    return r;

                if (workerCanExit()) {

                    if (runState >= SHUTDOWN) // Wake up others

                        interruptIdleWorkers();

                    return null;

                }

                // Else retry

            } catch (InterruptedException ie) {

                // On interruption, re-check runState

            }

        }

    }

 

//线程数多于corePoolSize或者allowCoreThreadTimeOut==true时，如果没有从队列中获取到任务（可能会等待到timeout）则会执行以下逻辑，判断是否需要退出该线程，符合退出的条件如下：

//1、线程池是STOP或者TERMINATED状态

//2、任务队列为空

//3、allowCoreThreadTimeOut==true并且线程数>corePoolSize？

private boolean workerCanExit() {

        final ReentrantLock mainLock = this.mainLock;

        mainLock.lock();

        boolean canExit;

        try {

            canExit = runState >= STOP ||

                workQueue.isEmpty() ||

                (allowCoreThreadTimeOut &&

                 poolSize > Math.max(1, corePoolSize));

        } finally {

            mainLock.unlock();

        }

        return canExit;

    }

 

  

评论

1 楼 flychao88 2014-05-06  

AbstractQueuedSynchronizer 核心就是用的这个，可以根据这个扩展来实现自己需要的