阻塞队列
LinkedBlockingQueue 是调用 Executors 类中的方法生成 ThreadPoolExecutor 实例时使用的默认队列，PriorityBlockingQueue 实际上也是一个BlockingQueue，不过，根据设定的优先级来处理任务也是一个棘手的问题。首先，提交一个 Runnable 或 Callable 任务，该任务被包装成一个 RunnableFuture，然后添加到队列中，ProrityBlockingQueue 比较每个对象来决定执行的优先权（比较对象是包装后的RunnableFuture而不是任务的内容）。不仅如此，当 corePoolSize 大于1并且工作线程空闲时，ThreadPoolExecutor 可能会根据插入顺序来执行，而不是 PriorityBlockingQueue 所期望的优先级顺序。
默认情况下，ThreadPoolExecutor 的工作队列（workQueue）是没有边界的。通常这是没问题的，但是请记住，没有边界的工作队列可能导致应用出现内存溢出（out of memory）错误。如果要限制任务队列的大小，可以设置 RejectionExecutionHandler。你可以自定义处理器或者从4个已有处理器（默认AbortPolicy）中选择一个：
CallerRunsPolicy
AbortPolicy
DiscardPolicy
DiscardOldestPolicy
线程工厂
线程工厂通常用于创建自定义的线程。例如，你可以增加自定义的 Thread.UncaughtExceptionHandler 或者设置线程名称。在下面的例子中，使用线程名称和线程的序号来记录未捕获的异常。
public class LoggingThreadFactory implements ThreadFactory {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private static final String THREAD_NAME_PREFIX = "worker-thread-";
private final AtomicInteger threadCreationCounter = new AtomicInteger();
@Override
public Thread newThread(Runnable task) {
int threadNumber = threadCreationCounter.incrementAndGet();
Thread workerThread = new Thread(task, THREAD_NAME_PREFIX + threadNumber);
workerThread.setUncaughtExceptionHandler(thread, throwable -> logger.error("Thread {} {}", thread.getName(), throwable));
return workerThread;
}
}
生产者消费者实例
生产者消费者是一种常见的同步多线程处理问题。在这个例子中，我们使用 ExecutorService 解决此问题。但是，这不是解决该问题的教科书例子。我们的目标是演示线程池来处理所有的同步问题，从而程序员可以集中精力去实现业务逻辑。
Producer 定期的从数据库获取新的数据来创建任务，并将任务提交给 ExecutorService。ExecutorService 管理的线程池中的一个工作线程代表一个 Consumer，用于处理业务任务（如计算价格并返回给客户）。
首先，我们使用 Spring 来配置：
@Configuration
public class ProducerConsumerConfiguration {
<a href='http://www.jobbole.com/members/weibo_1902876561'>@Bean</a>
public ExecutorService executorService() {
// single consumer
return Executors.newSingleThreadExecutor();
}
// other beans such as a data source, a scheduler, etc.
}
然后，建立一个 Consumer 及一个 ConsumerFactory。该工程方法通过生产者调用来创建一个任务，在未来的某一个时间点，会有一个工作线程执行该任务。
public class Consumer implements Runnable { private final BusinessTask businessTask; private final BusinessLogic businessLogic; public Consumer(BusinessTask businessTask, BusinessLogic businessLogic) { this.businessTask = businessTask; this.businessLogic = businessLogic; } @Override public void run() { businessLogic.processTask(businessTask); } } @Component public class ConsumerFactory { private final BusinessLogic businessLogic; public ConsumerFactory(BusinessLogic businessLogic) { this.businessLogic = businessLogic; } public Consumer newConsumer(BusinessTask businessTask) { return new Consumer(businessTask, businessLogic); } } |
最后，有一个 Producer 类，用于从数据库中获取数据并创建业务任务。在这个例子中，我们假定 fetchData() 是通过 scheduler 周期性调用的。
@Component public class Producer { private final DataRepository dataRepository; private final ExecutorService executorService; private final ConsumerFactory consumerFactory; @Autowired public Producer(DataRepository dataRepository, ExecutorService executorService, ConsumerFactory consumerFactory) { this.dataRepository = dataRepository; this.executorService = executorService; this.consumerFactory = consumerFactory; } public void fetchAndSubmitForProcessing() { List<Data> data = dataRepository.fetchNew(); data.stream() // create a business task from data fetched from the database .map(BusinessTask::fromData) // create a consumer for each business task .map(consumerFactory::newConsumer) // submit the task for further processing in the future (submit is a non-blocking method) .forEach(executorService::submit); } } |
非常感谢 ExecutorService，这样我们就可以集中精力实现业务逻辑，我们不需要担心同步问题。上面的演示代码只用了一个生产者和一个消费者。但是，很容易扩展为多个生产者和多个消费者的情况。
总结
JDK 5 诞生于2004年，提供很多有用的并发工具，ExecutorService 类就是其中的一个。线程池通常应用于服务器的底层（如 Tomcat 和 Undertow）。当然，线程池也不仅仅局限于服务器环境。在任何密集并行（embarrassingly parallel）难题中它们都非常有用。由于现在越来越多的软件运行于多核系统上，线程池就更值得关注了。
