ThreadPoolExecutor与ArrayBlockingQueue

Question

当我在我的一个项目中使用它时,我开始从Java Doc中阅读更多关于ThreadPoolExecutor的内容.所以任何人都可以解释一下这条线实际意味着什么吗？ - 我知道每个参数代表什么,但我想从这里的一些专家那里以更一般/非人的方式理解它.

ExecutorService service = new ThreadPoolExecutor(10, 10, 1000L,
TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(10, true), new 
ThreadPoolExecutor.CallerRunsPolicy());

更新: - 问题陈述是: -

每个线程使用1到1000之间的唯一ID,程序必须运行60分钟或更长时间,因此在60分钟内,所有ID都可能完成,所以我需要再次重用这些ID.所以这是我用以上执行程序编写的以下程序.

class IdPool {
    private final LinkedList<Integer> availableExistingIds = new LinkedList<Integer>();

    public IdPool() {
        for (int i = 1; i <= 1000; i++) {
            availableExistingIds.add(i);
        }
    }

    public synchronized Integer getExistingId() {
        return availableExistingIds.removeFirst();
    }

    public synchronized void releaseExistingId(Integer id) {
        availableExistingIds.add(id);
    }
}


class ThreadNewTask implements Runnable {
    private IdPool idPool;

    public ThreadNewTask(IdPool idPool) {
        this.idPool = idPool;
    }

    public void run() {
        Integer id = idPool.getExistingId();
        someMethod(id);
        idPool.releaseExistingId(id);
    }

// This method needs to be synchronized or not?
    private synchronized void someMethod(Integer id) {
        System.out.println("Task: " +id);
// and do other calcuations whatever you need to do in your program
    }
}

public class TestingPool {
    public static void main(String[] args) throws InterruptedException {
        int size = 10;
        int durationOfRun = 60;
        IdPool idPool = new IdPool();   
        // create thread pool with given size
        ExecutorService service = new ThreadPoolExecutor(size, size, 500L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(size), new ThreadPoolExecutor.CallerRunsPolicy()); 

        // queue some tasks
        long startTime = System.currentTimeMillis();
        long endTime = startTime + (durationOfRun * 60 * 1000L);

        // Running it for 60 minutes
        while(System.currentTimeMillis() <= endTime) {
            service.submit(new ThreadNewTask(idPool));
        }

        // wait for termination        
        service.shutdown();
        service.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS); 
    }
}

我的问题是: - 就考虑性能而言,此代码是正确的吗？还有什么我可以在这里使它更准确？任何帮助将不胜感激.

Answer 1

[首先,我道歉,这是对先前答案的回应,但我想要格式化].

除了在现实中,您不要阻止将项目提交给具有完整队列的ThreadPoolExecutor.原因是ThreadPoolExecutor调用BlockingQueue.offer(T item)方法,根据定义,该方法是一种非阻塞方法.它要么添加项目并返回true,要么不添加(填满时)并返回false.然后,ThreadPoolExecutor调用已注册的RejectedExecutionHandler来处理这种情况.

来自javadoc:

将来某个时候执行给定的任务.任务可以在新线程或现有池化线程中执行.如果无法提交执行任务,或者因为此执行程序已关闭或已达到其容量,则该任务由当前的RejectedExecutionHandler处理.

默认情况下,使用ThreadPoolExecutor.AbortPolicy(),它从ThreadPoolExecutor的"submit"或"execute"方法抛出RejectedExecutionException.

try {
   executorService.execute(new Runnable() { ... });
}
catch (RejectedExecutionException e) {
   // the queue is full, and you're using the AbortPolicy as the 
   // RejectedExecutionHandler
}

但是,您可以使用其他处理程序执行不同的操作,例如忽略错误(DiscardPolicy),或在调用"execute"或"submit"方法(CallerRunsPolicy)的线程中运行它.此示例允许在队列已满时调用"submit"或"execute"方法运行所请求任务的任何线程.(这意味着在任何给定的时间,你可以在池本身的内部运行另外一件事):

ExecutorService service = new ThreadPoolExecutor(..., new ThreadPoolExecutor.CallerRunsPolicy());

如果你想阻塞并等待,你可以实现你自己的RejectedExecutionHandler,它会阻塞,直到队列中有一个可用的插槽(这是一个粗略的估计,我没有编译或运行它,但你应该得到这个想法):

public class BlockUntilAvailableSlot implements RejectedExecutionHandler {
  public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
     if (e.isTerminated() || e.isShutdown()) {
        return;
     }

     boolean submitted = false;
     while (! submitted) {
       if (Thread.currentThread().isInterrupted()) {
            // be a good citizen and do something nice if we were interrupted
            // anywhere other than during the sleep method.
       }

       try {
          e.execute(r);
          submitted = true;
       }
       catch (RejectedExceptionException e) {
         try {
           // Sleep for a little bit, and try again.
           Thread.sleep(100L);
         }
         catch (InterruptedException e) {
           ; // do you care if someone called Thread.interrupt?
           // if so, do something nice here, and maybe just silently return.
         }
       }
     }
  }
}