Java中有界隊(duì)列的飽和策略(reject policy)原理解析

更新時(shí)間：2020年04月24日 10:26:52 作者：flydean程序那些事

這篇文章主要介紹了Java中有界隊(duì)列的飽和策略(reject policy)原理解析,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

我們?cè)谑褂肊xecutorService的時(shí)候知道，在ExecutorService中有個(gè)一個(gè)Queue來(lái)保存提交的任務(wù)，通過(guò)不同的構(gòu)造函數(shù)，我們可以創(chuàng)建無(wú)界的隊(duì)列（ExecutorService.newCachedThreadPool）和有界的隊(duì)列(ExecutorService newFixedThreadPool(int nThreads))。

無(wú)界隊(duì)列很好理解，我們可以無(wú)限制的向ExecutorService提交任務(wù)。那么對(duì)于有界隊(duì)列來(lái)說(shuō)，如果隊(duì)列滿了該怎么處理呢？

今天我們要介紹一下java中ExecutorService的飽和策略(reject policy)。

以ExecutorService的具體實(shí)現(xiàn)ThreadPoolExecutor來(lái)說(shuō)，它定義了4種飽和策略。分別是AbortPolicy，DiscardPolicy，DiscardOldestPolicy和CallerRunsPolicy。

如果要在ThreadPoolExecutor中設(shè)定飽和策略可以調(diào)用setRejectedExecutionHandler方法，如下所示：

    ThreadPoolExecutor threadPoolExecutor= new ThreadPoolExecutor(5, 10, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>(20));
    threadPoolExecutor.setRejectedExecutionHandler(
        new ThreadPoolExecutor.AbortPolicy()
    );

上面的例子中我們定義了一個(gè)初始5個(gè)，最大10個(gè)工作線程的Thread Pool，并且定義其中的Queue的容量是20。如果提交的任務(wù)超出了容量，則會(huì)使用AbortPolicy策略。

AbortPolicy

AbortPolicy意思是如果隊(duì)列滿了，最新的提交任務(wù)將會(huì)被拒絕，并拋出RejectedExecutionException異常：

  public static class AbortPolicy implements RejectedExecutionHandler {
    /**
     * Creates an {@code AbortPolicy}.
     */
    public AbortPolicy() { }

    /**
     * Always throws RejectedExecutionException.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     * @throws RejectedExecutionException always
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
      throw new RejectedExecutionException("Task " + r.toString() +
                         " rejected from " +
                         e.toString());
    }
  }

上面的代碼中，rejectedExecution方法中我們直接拋出了RejectedExecutionException異常。

DiscardPolicy

DiscardPolicy將會(huì)悄悄的丟棄提交的任務(wù)，而不報(bào)任何異常。

public static class DiscardPolicy implements RejectedExecutionHandler {
    /**
     * Creates a {@code DiscardPolicy}.
     */
    public DiscardPolicy() { }

    /**
     * Does nothing, which has the effect of discarding task r.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
  }

DiscardOldestPolicy

DiscardOldestPolicy將會(huì)丟棄最老的任務(wù)，保存最新插入的任務(wù)。

  public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    /**
     * Creates a {@code DiscardOldestPolicy} for the given executor.
     */
    public DiscardOldestPolicy() { }

    /**
     * Obtains and ignores the next task that the executor
     * would otherwise execute, if one is immediately available,
     * and then retries execution of task r, unless the executor
     * is shut down, in which case task r is instead discarded.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
      if (!e.isShutdown()) {
        e.getQueue().poll();
        e.execute(r);
      }
    }
  }

我們看到在rejectedExecution方法中，poll了最老的一個(gè)任務(wù)，然后使用ThreadPoolExecutor提交了一個(gè)最新的任務(wù)。

CallerRunsPolicy

CallerRunsPolicy和其他的幾個(gè)策略不同，它既不會(huì)拋棄任務(wù)，也不會(huì)拋出異常，而是將任務(wù)回退給調(diào)用者，使用調(diào)用者的線程來(lái)執(zhí)行任務(wù)，從而降低調(diào)用者的調(diào)用速度。我們看下是怎么實(shí)現(xiàn)的：

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    /**
     * Creates a {@code CallerRunsPolicy}.
     */
    public CallerRunsPolicy() { }

    /**
     * Executes task r in the caller's thread, unless the executor
     * has been shut down, in which case the task is discarded.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
      if (!e.isShutdown()) {
        r.run();
      }
    }
  }

在rejectedExecution方法中，直接調(diào)用了 r.run()方法，這會(huì)導(dǎo)致該方法直接在調(diào)用者的主線程中執(zhí)行，而不是在線程池中執(zhí)行。從而導(dǎo)致主線程在該任務(wù)執(zhí)行結(jié)束之前不能提交任何任務(wù)。從而有效的阻止了任務(wù)的提交。

使用Semaphore

如果我們并沒(méi)有定義飽和策略，那么有沒(méi)有什么方法來(lái)控制任務(wù)的提交速度呢？考慮下之前我們講到的Semaphore，我們可以指定一定的資源信號(hào)量來(lái)控制任務(wù)的提交，如下所示：

public class SemaphoreUsage {

  private final Executor executor;
  private final Semaphore semaphore;

  public SemaphoreUsage(Executor executor, int count) {
    this.executor = executor;
    this.semaphore = new Semaphore(count);
  }

  public void submitTask(final Runnable command) throws InterruptedException {
    semaphore.acquire();
    try {
      executor.execute(() -> {
            try {
              command.run();
            } finally {
              semaphore.release();
            }
          }
      );
    } catch (RejectedExecutionException e) {
      semaphore.release();
    }
  }
}

本文的例子可參考https://github.com/ddean2009/learn-java-concurrency/tree/master/rejectPolicy

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: