java - 带 ArrayBlockingQueue 的 ThreadPoolExecutor

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:

Executes the given task sometime in the future. The task may execute in a new thread or in an existing pooled thread. If the task cannot be submitted for execution, either because this executor has been shutdown or because its capacity has been reached, the task is handled by the current RejectedExecutionHandler.

默认情况下，使用 ThreadPoolExecutor.AbortPolicy() 从 ThreadPoolExecutor 的“提交”或“执行”方法中抛出 RejectedExecutionException。

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

但是，您可以使用其他处理程序来做一些不同的事情，例如忽略错误(DiscardPolicy)，或者在调用“执行”或“提交”方法(CallerRunsPolicy)的线程中运行它。此示例让调用“提交”或“执行”方法的线程在队列已满时运行请求的任务。 (这意味着在任何给定时间，您都可以在池本身的内容之上运行 1 个额外的东西):

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