是否可以向ThreadPoolExecutor的BlockingQueue添加任务？

5 回答

• 4

  通过在实例化时指定 RejectedExecutionHandler ，实际上可以在队列满时配置池的行为 . ThreadPoolExecutor 将四个策略定义为内部类，包括 AbortPolicy ， DiscardOldestPolicy ， DiscardPolicy ，以及我个人最喜欢的 CallerRunsPolicy ，它在控制线程中运行新作业 .

  例如：

  ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
          nproc, // core size
          nproc, // max size
          60, // idle timeout
          TimeUnit.SECONDS,
          new ArrayBlockingQueue<Runnable>(4096, true), // Fairness = true guarantees FIFO
          new ThreadPoolExecutor.CallerRunsPolicy() ); // If we have to reject a task, run it in the calling thread.
  

  可以使用以下内容获得问题中所需的行为：

  public class BlockingPolicy implements RejectedExecutionHandler {
      void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
          executor.getQueue.put(r); // Self contained, no queue reference needed.
      }
  

  在某些时候，必须访问队列 . 这样做的最佳位置是自包含的 RejectedExecutionHandler ，它可以保存在池对象范围内直接操作队列所产生的任何代码重复或潜在错误 . 请注意， ThreadPoolExecutor 中包含的处理程序本身使用 getQueue() .

  回复于 2023-09-27T08:11:30+00:00
• 0

  如果是我，我宁愿使用 Executor#execute() 而不是 Queue#offer() ，因为我已经使用了 java.util.concurrent 中的所有其他内容 .

  你的问题很好，引起了我的兴趣，所以我看了 ThreadPoolExecutor#execute() 的来源：

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

  我们可以看到execute本身在工作队列中调用了 offer() ，但是在必要之前没有做一些漂亮，美味的池操作 . 因此，我建议使用 execute() ;不使用它可能（虽然我认为使用 offer() 会破坏执行程序 - 看起来任务是使用以下（也来自ThreadPoolExecutor）从队列中拉出的：

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

  这个 getTask() 方法只是在循环中调用，所以如果执行程序's not shutting down, it'阻塞，直到给队列一个新任务（无论它来自哪里） .

  注意：尽管我依赖它们来获得明确的答案 - 我们应该只编写API . 我们不知道 execute() 的实施将如何随着时间而改变 .

  回复于 2023-09-27T08:11:30+00:00
• 8

  一个技巧是实现ArrayBlockingQueue的自定义子类并覆盖offer（）方法来调用阻塞版本，然后您仍然可以使用正常的代码路径 .

  queue = new ArrayBlockingQueue<Runnable>(queueSize) {
    @Override public boolean offer(Runnable runnable) {
      try {
        return offer(runnable, 1, TimeUnit.HOURS);
      } catch(InterruptedException e) {
        // return interrupt status to caller
        Thread.currentThread().interrupt();
      }
      return false;
    }
  };
  

  （正如你可能猜到的那样，我认为直接在队列中调用offer是正常的代码路径可能是一个坏主意） .

  回复于 2023-09-27T08:11:30+00:00
• 11

  它使用的是与标准内存 LinkedBlockingQueue 或 ArrayBlockingQueue 完全不同的实现 .

  例如，如果你想要像OP一样阻止 offer() .

  因此，给定的答案是，必须调用 prestartAllCoreThreads() （或足够的时间 prestartCoreThread() ）以使工作线程可用并运行，这一点非常重要 .

  回复于 2023-09-27T08:11:30+00:00
• 1

  如果需要，我们还可以使用一个停车场，将主要处理与被拒绝的任务分开 -

  final CountDownLatch taskCounter = new CountDownLatch(TASKCOUNT);
      final List<Runnable> taskParking = new LinkedList<Runnable>();
      BlockingQueue<Runnable> taskPool = new ArrayBlockingQueue<Runnable>(1);
      RejectedExecutionHandler rejectionHandler = new RejectedExecutionHandler() {
          @Override
          public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
              System.err.println(Thread.currentThread().getName() + " -->rejection reported - adding to parking lot " + r);
              taskCounter.countDown();
              taskParking.add(r);
          }
      };
      ThreadPoolExecutor threadPoolExecutor =  new ThreadPoolExecutor(5, 10, 1000, TimeUnit.SECONDS, taskPool, rejectionHandler);
      for(int i=0 ; i<TASKCOUNT; i++){
          //main 
          threadPoolExecutor.submit(getRandomTask());
      }
      taskCounter.await(TASKCOUNT * 5 , TimeUnit.SECONDS);
      System.out.println("Checking the parking lot..." + taskParking);
      while(taskParking.size() > 0){
          Runnable r = taskParking.remove(0);
          System.out.println("Running from parking lot..." + r);
          if(taskParking.size() > LIMIT){
            waitForSometime(...);
          }
          threadPoolExecutor.submit(r);
      }
      threadPoolExecutor.shutdown();
  

  回复于 2023-09-27T08:11:30+00:00