更改ThreadPoolExecutor

Question

当一个任务要执行时，它会被插入到队列中，当队列返回false时，将为其执行创建一个新线程。当前位于池中的空闲线程不会被拾取。原因是任务从getTask()方法分配给空闲线程，该方法从队列中选择任务。所以我的问题是如何改变这种行为，以便如果线程是空闲的，如何确保为空闲线程分配执行任务，而不是创建新线程？？

下面的输出将使其更加清楚：-

Task 46 ends
Active Count: 0 Pool Size : 3 Idle Count: 3 Queue Size: 0
Task 47 ends
Active Count: 0 Pool Size : 3 Idle Count: 3 Queue Size: 0
Task 48 ends
Active Count: 0 Pool Size : 3 Idle Count: 3 Queue Size: 0
Active Count: 1 Pool Size : 4 Idle Count: 3 Queue Size: 0
Task 49 ends
Active Count: 2 Pool Size : 5 Idle Count: 3 Queue Size: 0
Task 50 ends
Active Count: 2 Pool Size : 5 Idle Count: 3 Queue Size: 0

代码文件如下：-

 public void execute(Runnable command) {
    System.out.println("Active Count: " + getActiveCount()
            + " Pool Size : " + getPoolSize() + " Idle Count: "
            + (getPoolSize() - getActiveCount())+" Queue Size: "+getQueue().size());
          if (command == null)
              throw new NullPointerException();
          for (;;) {
              if (runState != RUNNING) {
                  reject(command);
                  return;
              }
              if (poolSize < corePoolSize && addIfUnderCorePoolSize(command))
                  return;
              if (workQueue.offer(command))
                  return;
              int status = addIfUnderMaximumPoolSize(command);
              if (status > 0)      // created new thread
                  return;
              if (status == 0) {   // failed to create thread
                  reject(command);
                  return;
              }
              // Retry if created a new thread but it is busy with another task
          }
      }

public class CustomBlockingQueue<E> extends LinkedBlockingQueue<E> 
{

    /**
     * 
     */
    private static final long serialVersionUID = 1L;

    public CustomBlockingQueue() {
    super(Integer.MAX_VALUE);
    }

    public boolean offer(E e) {
       return false;
    }

}

   final CustomThreadPoolExecutor tpe = new CustomThreadPoolExecutor(3, 8, 0L, TimeUnit.MILLISECONDS, new MediationBlockingQueue<Runnable>(), new MediationRejectionHandler());

private static final int TASK_COUNT = 100;

 for (int i = 0; i < TASK_COUNT; i++) {

......
tpe.execute(new Task(i));
.....
}

我们调用的执行器的核心池大小为3，最大池大小为8，并对任务使用无界链接阻塞队列。

Answer 1

使用SynchronousQueue实现“在排队前开始但首选现有线程”行为的最简单方法。当且仅当已经有一个等待接收者时，它将接受提供的项目。因此，空闲线程将获得项目，一旦没有空闲线程，ThreadPoolExecutor将启动新的线程。

唯一的缺点是，一旦启动了所有线程，就不能简单地将挂起的项放入队列，因为它没有容量。因此，您要么接受提交器被阻塞，要么需要另一个队列将挂起的任务放入它，另一个后台线程将这些挂起的项目放入同步队列。这个额外的线程不会损害性能，因为它大部分时间都被阻塞在这两个队列中的任何一个中。

class QueuingRejectionHandler implements RejectedExecutionHandler {

  final ExecutorService processPending=Executors.newSingleThreadExecutor();

  public void rejectedExecution(
      final Runnable r, final ThreadPoolExecutor executor) {
    processPending.execute(new Runnable() {
      public void run() {
        executor.execute(r);
      }
    });
  }
}

…

ThreadPoolExecutor e=new ThreadPoolExecutor(
  corePoolSize, maximumPoolSize, keepAliveTime, unit,
  new SynchronousQueue<Runnable>(), new QueuingRejectionHandler());