限制Java線程池運(yùn)行線程以及等待線程數(shù)量的策略

Clect 發(fā)布于2019-08-14 14:13 / 1969人閱讀

摘要：限制線程池運(yùn)行線程以及等待線程數(shù)量的策略對(duì)于所提供的，可以保證可以在內(nèi)存中有固定數(shù)量的線程數(shù)運(yùn)行。指的是當(dāng)線程池拒絕該任務(wù)的時(shí)候，線程在本地線程直接。由此限制了線程池的等待線程數(shù)與執(zhí)行線程數(shù)

限制Java線程池運(yùn)行線程以及等待線程數(shù)量的策略

對(duì)于java.util.concurrent.Executors所提供的FixedThreadPool，可以保證可以在內(nèi)存中有固定數(shù)量的線程數(shù)運(yùn)行。但是由于FixedThreadPool綁定的是LinkedBlockingQueue。隊(duì)列的上限沒(méi)有限制（默認(rèn)上限為Integer.MAX_VALUE），不斷的提交新的線程，會(huì)造成任務(wù)在內(nèi)存中長(zhǎng)時(shí)間的堆積。

我們有可能面臨如下的場(chǎng)景，主線程不斷地提交任務(wù)線程，希望有固定數(shù)量的在線程中運(yùn)行，也不想造成線程在內(nèi)存中大量的等待堆積。由此需要我們自己定義一個(gè)線程池策略。ThreadPoolExecutor為我們線程池的設(shè)置提供了很大的靈活性。

首先看FixedThreadPool的實(shí)現(xiàn)：

    public static ExecutorService More ...newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
            return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue(),
                                          threadFactory);
        }

可以看到，F(xiàn)ixedThreadPool綁定的是LinkedBlockingQueue。我們需要做的第一個(gè)改造就是綁定有大小上線的BlockingQueue，在我的實(shí)現(xiàn)中綁定ArrayBlockingQueue并設(shè)置了size。

第二個(gè)是采用CallerRunsPolicy。ThreadPoolExecutor可以定義不同的任務(wù)拒絕策略。CallerRunsPolicy指的是當(dāng)線程池拒絕該任務(wù)的時(shí)候，線程在本地線程直接execute。這樣就限制了本地線程的循環(huán)提交流程。

    BlockingQueue workingQueue = new ArrayBlockingQueue(10);
    RejectedExecutionHandler rejectedExecutionHandler =
        new ThreadPoolExecutor.CallerRunsPolicy();
    ExecutorService threadPool = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS,
        workingQueue, rejectedExecutionHandler);

    for (int i = 0; i < 100; i++) {
      
      threadPool.submit(new Callable() {

        @Override
        public Boolean call() throws Exception {
          System.out.println("thread " + String.valueOf(threadNo) + " is called");
          Thread.sleep(10000);
          System.out.println("thread " + String.valueOf(threadNo) + " is awake");
          throw new Exception();
        }

      });
    }

代碼中定義了大小為10的線程池，for循環(huán)提交了20個(gè)線程的時(shí)候，10個(gè)執(zhí)行線程，10個(gè)線程放入了workingQueue。當(dāng)提交到第21個(gè)線程的時(shí)候，會(huì)觸發(fā)RejectedExecutionHandler。在這里我們配置了CallerRunsPolicy策略。所以會(huì)在主線程直接執(zhí)行該線程。也就是說(shuō)，在本程序中最多會(huì)有11個(gè)線程在執(zhí)行，10個(gè)線程在等待。由此限制了線程池的等待線程數(shù)與執(zhí)行線程數(shù)