大多数服务端应用程序都需要同时处理任务的能力这样可以提高工作性能并增加硬件资源的利用在早期的Java版本(或更早的)中开发者需要完成并发(concurrent)应用程序——包括线程池逻辑—他们自己使用的是低层次语言结构和Java Thread API但是结果却总是不理想Java Thread API的特性会导致不知情的编程者开发一些难以调试的编程错误的代码
在Java中Sun公司采用了Java concurrency功能(JSR)来解决这些问题并且提供了一套标准的APIs来创建并开发应用程序本文探究了一些Java concurrency package提供的特性并使用这些功能来演示编写并发应用程序的技术
Concurrent Programming的挑战
自从它发布以来Java就提供了Thread类和低层次语言结构例如synchronized 和volatile用来开发独立平台的并发应用程序但是用这些特性来构建并发应用程序并不是简单的事情开发者要面对以下的挑战
不正确的编程会导致一些困境就是两个或两个以上的线程都等待永远被锁住的对方
在Java语言中没有机制用于写waitfree lockfree算法开发者必须使用本地代码
开发者必须编写他们复杂的线程池逻辑这样会很棘手并且容易出错
Java Concurrency Utilities的概述
JSR(Java concurrency utilities)是Java的一部分通过着重在宽度并提供跨域大范围并发编程风格的重要功能大大简化了在Java中并发应用程序的开发
Java concurrency utilities提供了多种功能开发者可以应用于更快更有预见性的并发应用程序的开发这些功能让开发者从通过写自定义代码来重新发明wheel中解放出来一些JSR的最显着的特点是
标准的接口和构架来定义自定义线程子系统
是一种机制用于规范调用时序安排执行和异步任务的控制这是根据在Executor构架中的一套执行政策
是一种机制通过类用于线程协调例如semaphore mutexe barrier latche和 exchangers
非阻碍FIFO列队执行(ConcurrentLinkedQueue类)用于可升级的有效的线程安全操作
阻碍列队执行类来涵盖最常见的使用情况对于生成者/消费者方法信息并行任务和相关的并发设计
是一种构架用于锁定和等待不同于内置同步和监测器的条件
随时准备使用的类用于在单个变量上的锁定自由线程安全的编程
开发一个Concurrent Java Application
本节是演示如何使用Java concurrency utility API来开发一个多线程的在线订单程序的电子商务应用程序在应用程序生效并授权命令之后把它们放在订单处理列队(ncurrentBlockingQueue)订单处理器线程池不断的对订单进行测验而且当这些订单可以使用时进行处理
解耦应用程序的订单处理代码提供了增加和减少订单处理率的灵活性通过改变线程池的大小在一个并发BlockingQueue中放入订单对象确保一个处理器处理一个订单而且它要照顾自动同步
在以下小节中的代码段是截取于伴随本文中的应用程序源代码
扩展ThreadPoolExecutor
Executor接口只规定了一个方法并从任务如何运行中解耦任务提交ExecutorService子接口规定了额外的方法用于提交并追蹤异步任务以及关闭线程池ThreadPoolExecutor类是ExecutorService接口的一个具体的执行应该足以用于大多数订单处理应用程序的需求
ThreadPoolExecutor也提供有用的连接方法(eg beforeExecute)可以覆盖定制目的在Listing 中的CustomThreadPoolExecutor类扩展了ThreadPoolExecutor类并覆盖了beforeExecute afterExecute和 terminated 方法
@Override
public void afterExecute(Runnable r Throwable t) {
superafterExecute(r t);
Loggerlog(After calling afterExecute() method for a thread
+ r);
}
@Override
public void terminated() {
superterminated();
Loggerlog(Threadpool terminated);
}
该覆盖方法只需登陆复写信息
但是在现实的情况中
它们能做更有用的事情
例如
terminated方法可以发送一个在这个池中的所有线程都死锁的警告
要正确构建多重覆盖
通常你应该从在子类中的各个方法中调用主类的覆盖方法
Java concurrency API还提供了ScheduledExecutorService接口可以扩展ExecutorServiceInterface而且在一个特定延迟或是定期的执行之后能够安排任务进行运行ScheduledExecutorThreadPool是这个接口的具体执行
确定异步任务执行
Executor 执行提交异步任务这些任务执行实际的业务逻辑向executor提交一个任务ExecutorService接口提供重载的submit方法可以接受Runnable 或是Callable 对象类型
Runnable任务类型在以下情况下市非常有用的
任务完成时不需要返回任何结果
如果run()方法遇到一个例外没有必要抛出一个特定程序检查例外
移植现有的legacy 类来实施Runnable接口是必需的
Callable任务类型提供了更多灵活性并提供了下列的优点
任务可以返回用户定义对象作为结果
任务可以抛出用户定义的检查例外
你需要分别为Runnable 和Callable任务类型执行run() 和call()方法
在Listing 中的OrderProcessorCallable类执行Callable接口并指定一个Integer作为结果对象Constructor把任务对象名称和BlockingQueue当做检索命令来处理call()方法继续为订单值对象调查BlockingQueue并且处理任何所它所发现的事情如果没有订单处理call()方法会休息一段时间再继续工作
Call方法的无限循环在这个应用程序方案中是非常有用的因为因为没有必要一次又一次为每个订单对象去创建并提交新的任务到ThreadPoolExecutor中
public Integer call() throws OrderProcessingException {
while (running) {
// check if current Thread is interrupted
checkInterruptStatus();
// poll for OrderVO from blocking queue and do
// order processing here
}
// return result
return processedCount;
}
请注意异步任务执行不断的为应用程序的生命周期运行
在大多数程序中
异步任务执行会进行必要的操作并立即返回
处理线程中断
Call方法执行使用checkInterruptStatus方法在执行线程中断上进行经常性检查这是必需的因为为了迫使任务取消ThreadPoolExecutor会向线程发送一个interrupt否则检查中断状态会导致特定线程再也无法返回以下的checkInterruptStatus方法检查运行线程的中断状态如果线程被中断会抛出OrderProcessingException
private void checkInterruptStatus() throws
OrderProcessingException {
if (Threadinterrupted()) {
throw new OrderProcessingException(Thread was interrupted);
}
}
作为任务的实施
它是抛出一个异常并在运行线程被中断的时候终止任务执行的一个非常好的练习
但是
基于订单处理的应用程序需求
在这种情况下忽略线程中断是非常恰当的
