如何使用纯ThreadPoolExecutor获取MoreExecutors.newDirectExecutorService()行为?

问题描述:

当我运行下面的代码:如何使用纯ThreadPoolExecutor获取MoreExecutors.newDirectExecutorService()行为?

package foo.trials; 

import com.google.common.util.concurrent.MoreExecutors; 
import org.slf4j.Logger; 
import org.slf4j.LoggerFactory; 

import java.util.Random; 
import java.util.concurrent.Callable; 
import java.util.concurrent.ExecutionException; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Future; 
import java.util.concurrent.Semaphore; 
import java.util.concurrent.SynchronousQueue; 
import java.util.concurrent.ThreadPoolExecutor; 
import java.util.concurrent.TimeUnit; 

public class DirectExecutorService { 
    private static final Logger logger_ = LoggerFactory.getLogger(DirectExecutoService.class); 

    public static void main(String[] args) { 
     boolean useGuava = true; 

     final ExecutorService directExecutorService; 
     if (useGuava) { 
      directExecutorService = MoreExecutors.newDirectExecutorService(); 
     } else { 
      directExecutorService = new ThreadPoolExecutor(
        0, 1, 0, TimeUnit.DAYS, 
        new SynchronousQueue<Runnable>(), 
        new ThreadPoolExecutor.CallerRunsPolicy()); 
      directExecutorService.submit(new BlockingCallable()); 
     } 

     Future<Boolean> future = directExecutorService.submit(new MyCallable()); 
     try { 
      logger_.info("Result: {}", future.get()); 
     } catch (InterruptedException e) { 
      logger_.error("Unexpected: Interrupted!", e); 
     } catch (ExecutionException e) { 
      logger_.error("Unexpected: Execution exception!", e); 
     } 
     logger_.info("Exiting..."); 
    } 

    static class MyCallable implements Callable<Boolean> { 
     static final Random _random = new Random(); 
     @Override 
     public Boolean call() throws Exception { 
      logger_.info("In call()"); 
      return _random.nextBoolean(); 
     } 
    } 

    static class BlockingCallable implements Callable<Boolean> { 

     Semaphore semaphore = new Semaphore(0); 
     @Override 
     public Boolean call() throws Exception { 
      semaphore.acquire(); // this will never succeed. 
      return true; 
     } 
    } 
} 

我得到以下输出

13:36:55.960 [main] INFO a.t.DirectExecutoService - In call() 
13:36:55.962 [main] INFO a.t.DirectExecutoService - Result: true 
13:36:55.963 [main] INFO a.t.DirectExecutoService - Exiting... 

注意,所有的执行发生在main线程。特别是可调用的调用get在调用线程中调度。当然,这是从MoreExecutors.newDirectExecutorService()预计那里不会有什么惊喜。

当我将变量useGuava设置为false时,我得到了类似的结果。

13:45:14.264 [main] INFO a.t.DirectExecutoService - In call() 
13:45:14.267 [main] INFO a.t.DirectExecutoService - Result: true 
13:45:14.268 [main] INFO a.t.DirectExecutoService - Exiting... 

,如果我注释掉以下行

directExecutorService.submit(new BlockingCallable()); 

然后我得到以下输出。

13:37:27.355 [pool-1-thread-1] INFO a.t.DirectExecutoService - In call() 
13:37:27.357 [main] INFO a.t.DirectExecutoService - Result: false 
13:37:27.358 [main] INFO a.t.DirectExecutoService - Exiting... 

正如人们所看到的可调用的调用发生在不同的线程pool-1-thread-1。我想我可以解释为什么会发生这种情况。也许因为线程池可以有(最多)1个可用的线程,所以第一个Callable被分派到该额外线程,否则该线程被BlockingCallable消耗。

我的问题是,如何创建一个ExecutorService,将DirectExecutorService做什么,而不必人为地燃烧线程与永不完成的可调用?

为什么我问这个?

  1. 我有一个在版本11.0使用番石榴的代码库。我需要避免将其升级到17.0+ - 如果可以的话,它提供MoreExecutors.newDirectExecutorService()
  2. ThreadPoolExecutor不允许将maxThreads设置为0.如果允许,那将会很奇怪,但如果确实如此,那也能解决我的问题。
  3. 最后,我很惊讶地发现这种行为 - 我本以为(错误地),使用CallerRunsPolicy会导致call所有Callable所有s到在调用者的线程中执行。所以,我想把我的经验和窍门放在那里,这样别人就可以节省最后燃烧的时间来试图理解这一点。 :(

有没有更好/更地道的方式来实现DirectExecutorService样的行为如果不能升级到番石榴17.0+?

有没有更好/更地道的方式来实现DirectExecutorService样的行为,如果一个不能升级到番石榴17.0+?

如果这是你唯一的问题在这里,你应该使用MoreExecutors.sameThreadExecutor()。它基本上newDirectExecutorService()之前,它被转移到一个新的方法(和加入directExecutor()),见Javadoc

由于:18.0(目前为MoreExecutors.sameThreadExecutor()自10.0)

BTW:你应该升级到最新的番石榴,你正在使用几乎六岁的一个!

+0

就是这样,谢谢!关于升级的观点很好。 –