ThreadPoolExecutor和CyclicBarrier配合使用可能带来的隐患

今天是很蛋疼的一天，排查一个bug排查了4个多小时。

情形简化之后大概是这样的：

我使用了spring的ThreadPoolTaskExecutor来进行并发时候的异步处理。并且给任务Runnable加上了CyclicBarrier，以达到让所有线程处理完之后再进行主线程的下一步操作的目的。其中executor的配置如下：

	<bean id="coreBlockExecutor"
		class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
		<property name="corePoolSize" value="5" />
		<property name="maxPoolSize" value="2000" />
		<property name="queueCapacity" value="1000" />
	</bean>

结果，bug就这样华丽丽的出现了：我new出来了1000个runnable，但无论如何也只能执行出5个runnable的结果（即corePoolSize），其他runnable就一直呆在blockingqueue里面不动弹了。


**************************************************** 华丽的分割线——原因 ****************************************************

花了n久去查看ThreadPoolTaskExecutor和ThreadPoolExecutor的源码。
发现是这样的，其实ThreadPoolTaskExecutor也就是调用了ThreadPoolExecutor，在后者中，有这样一个方法：

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
            if (runState == RUNNING && workQueue.offer(command)) {
                if (runState != RUNNING || poolSize == 0)
                    ensureQueuedTaskHandled(command);
            }
            else if (!addIfUnderMaximumPoolSize(command))
                reject(command); // is shutdown or saturated
        }
    }

可以看到，在当前线程池的线程数大于等于corePoolSize时候，会判断

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

也就是，当前线程池是处于运行状态并且队列是否还能插入runnable。
所以，当队列满的时候，也就是会去执行下面的代码

 else if (!addIfUnderMaximumPoolSize(command))
                reject(command); // is shutdown or saturated

即当前线程数如果小于最大线程数，则会调用addThread方法，将runnbale实例化成一个内部类Worker，加入线程池中运行。代码如下：

    private Thread addThread(Runnable firstTask) {
        Worker w = new Worker(firstTask);
        Thread t = threadFactory.newThread(w);
        if (t != null) {
            w.thread = t;
            workers.add(w);
            int nt = ++poolSize;
            if (nt > largestPoolSize)
                largestPoolSize = nt;
        }
        return t;
    }

而普通情况，即使填入executor的runnable数量不能填满queue，但在核心线程中运行的任务Worker结束之后，过了最大空闲时间（keepAliveTime）之后，即会释放线程，去从queue中获取等待中的任务。

代码如下：

        public void run() {
            try {
                Runnable task = firstTask;
                firstTask = null;
                while (task != null || (task = getTask()) != null) {
                    runTask(task);
                    task = null;
                }
            } finally {
                workerDone(this);
            }
        }
    }

    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
            }
        }
    }

原来Worker中的run方法是通过轮询来获取当前任务是否结束以及从队列里面继续拿任务的。当符合以下其中一个条件时：
可以看到，只有Worker当前的任务已完成的时候，才能去队列里面拿其他任务。否则，就只能keep on waiting了



所以原因就出来了：使用了CyclicBarrier进行栅栏式的改装之后，所有核心线程中的任务都会一直等待，不会空闲下来。这样核心线程就永远处于尴尬的被hold住的状态了。既不能结束当前的任务，也无法从队列获取新的任务，更无法中止线程了。


所以，像我这样没有仔细看源码的码农，碰到问题就悲剧了。哎




**************************************************** 华丽的分割线——解决方法 ****************************************************

1、允许的情况下，barrier的await设置过期时间
2、仔细考虑queue长度和并发规模

总之，好像也没什么特别好的办法。

评论

2 楼 Surlymo 2012-08-02  

lijunlong 写道

用CountDownLatch呢

这个一样的吧。

1 楼 lijunlong 2012-08-01  

用CountDownLatch呢