(转)Spring提供的线程池支持

一旦企业应用越来越复杂时（比如，基于流程服务器的EIS），它们对相关技术也提出了更高的要求。在使用EJB 3.0组件技术开发企业应用过程中，它们能够享受到EJB容器提供的线程池、任务调度（@Timeout）服务。现如今，运行于Web容器的Web应用、单独的桌面应用也复杂到需要依赖于线程池、任务调度的这类服务，是时候实现贵族到平民的转变了。

过去，很多企业项目可能会自身实现这类底层的非功能性的服务。从Java SE 5.0开始，线程池服务（即，java.util.concurrent包）已经内置到JDK中。至于任务调度服务，其实，自从Java 2 SDK 1.3以来，Java 2就内置了用于任务调度的定时器（比如，java.util.Timer、javax.swing.Timer）。开源领域的Quartz Scheduler正是能够提供企业级任务调度服务的使能技术，而且EJB 2.x也加强了任务调度的支持。甚至，BEA同IBM合作开发了Timer and Work Manager for Application Servers技术规范（CommonJ），这是专门用来解决Java EE应用中的线程池、任务调度问题的。另外，JMX规范中也定义了javax.management.timer.TimerMBean，开发者借助于它能够实现定时JMX通知。

针对上述各种线程池、任务调度支持，Spring 2.0提供了统一的客户视图、抽象，这使得应用根本不用理会底层的具体实现和机制。本章将从分析Spring 2.0提供的线程池支持入手，并过渡到Spring 2.0对任务调度提供的支持当中，从而进入到下一章内容。


自从Spring 2.0开始，TaskExecutor接口被引入到Spring平台中，这主要受到Java SE 5.0中java.util.concurrent.Executor的影响。这一接口为各种线程池服务提供了抽象，它在统一客户视图方面起到了最重要的作用。无论是Spring 2.0内部实现中，还是各种基于Spring的企业应用，TaskExecutor的应用随处可见，其定义如下。

public interface TaskExecutor {
  //异步或同步执行用户提交的任务
  void execute(Runnable task);
}

开发者可以通过execute(Runnable task)方法将待执行的任务提交给TaskExecutor。依据不同的TaskExecutor接口实现，这一任务会以异步或同步的方式进行。如果是同步，则调用者一直处于阻塞状态，直到任务被执行完成。此时，调用者同目标任务的执行处于同一线程中，因此线程上下文信息能够传播到目标任务的执行过程中。如果是异步，则一旦提交完任务，调用者即可返回，并继续进行自身的其他操作。此时，调用者同目标任务的执行位于不同的线程中，因此线程上下文信息很可能不能够在它们之间共享。应用要合理选择同步或异步。比如，在调用execute()期间，如果采用Spring受管事务或Acegi提供的企业级安全性服务，则一旦同步或异步选用不当，事务的ACID属性将得不到保证，而且应用的安全性也得不到保障。

Spring 2.0内置的TaskExecutor接口实现见图15-1，图中除了SyncTaskExecutor外，其他实现都是采用异步方式执行提交的任务的。

开发者是否还记得，第5章介绍的SimpleApplicationEventMulticaster，它使用了SyncTaskExecutor辅助完成事件的消费工作，其代码摘录如下。

public class SyncTaskExecutor implements TaskExecutor, Serializable {
  //同步调用客户提交的任务
   public void execute(Runnable task){
   Assert.notNull(task, "Runnable must not be null");
   task.run();
  }
} 

SyncTaskExecutor的使用非常简单，下面给出了示例，摘自Eclipse taskexecutordemo项目。

TaskExecutor te = new SyncTaskExecutor();
te.execute(new LogRunner());

由于SyncTaskExecutor不存在任何属性，因此它的使用最简单。在使用这一实现的过程中，并不会处触发新线程的创建工作，因为提交的任务同调用者同处于一个线程中。相比之下，异步执行用户任务的SimpleAsyncTaskExecutor的使用复杂些，下面给出了示例代码。此时，它会启动新的Thread，并允许开发者控制并发线程的上限（concurrencyLimit），从而起到一定的资源节流作用。默认时，concurrencyLimit取值为–1，即不启用资源节流。

<bean id="simpleAsyncTaskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor">
<property name="daemon" value="true" />
<property name="concurrencyLimit" value="2" />
<property name="threadNamePrefix" value="simpleAsyncTaskExecutor" />
</bean>

在Spring 2.0内部实现中，JMS集成会使用SimpleAsyncTaskExecutor完成JMS消息监听器的注册、执行等工作，我们将在第16章介绍这方面的内容。每次用户提交新的任务给SimpleAsyncTaskExecutor时，它都会启动新的线程来响应客户请求。这意味着，它并没有提供线程池功能。更何况，在Java EE环境中，随便启动新的线程并不是推荐的做法，因为Java EE容器没有办法管理到这些线程。所以，我们要改进SimpleAsyncTaskExecutor。

Java SE 5.0引入了ThreadPoolExecutor、ScheduledThreadPoolExecutor。Spring 2.0借助于ConcurrentTaskExecutor和ThreadPoolTaskExecutor能够通过IoC配置形式自定义它们暴露的各个属性。比如，下面给出ThreadPoolTaskExecutor的使用示例，其暴露的各个属性其实是ThreadPoolExecutor的属性，这体现了DI容器的优势。

<bean id="threadPoolTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="corePoolSize" value="2" />
<property name="keepAliveSeconds" value="200" />
<property name="maxPoolSize" value="10" />
<property name="queueCapacity" value="60" />
</bean>

<bean id="threadPoolTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">           
	    <!-- 核心线程数，默认为1 -->           
	    <property name="corePoolSize" value="10" />           
	    <!-- 最大线程数，默认为Integer.MAX_VALUE -->           
	    <property name="maxPoolSize" value="10" />           
	    <!-- 队列最大长度，一般需要设置值>=notifyScheduledMainExecutor.maxNum；默认为Integer.MAX_VALUE -->           
	    <property name="queueCapacity" value="60" />           
	    <!-- 线程池维护线程所允许的空闲时间，默认为60s -->           
	    <property name="keepAliveSeconds" value="200" />           
	    <!-- 线程池对拒绝任务（无线程可用）的处理策略，目前只支持AbortPolicy、CallerRunsPolicy；默认为后者 -->           
	    <property name="rejectedExecutionHandler">           
	        <!-- AbortPolicy:直接抛出java.util.concurrent.RejectedExecutionException异常 -->           
	        <!-- CallerRunsPolicy:主线程直接执行该任务，执行完之后尝试添加下一个任务到线程池中，可以有效降低向线程池内添加任务的速度 -->           
	        <!-- DiscardOldestPolicy:抛弃旧的任务、暂不支持；会导致被丢弃的任务无法再次被执行 -->           
	        <!-- DiscardPolicy:抛弃当前任务、暂不支持；会导致被丢弃的任务无法再次被执行 -->           
	        <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" />           
	    </property>
    </bean>

Spring 2.0在集成java.util.Timer任务调度机制过程中，还引入了TimerTaskExecutor。此时，用户提交的所有任务将交由Timer本身的单个线程执行，其示例配置如下。这一单个线程会以串行的方式完成用户提交给它的所有任务。

<bean id="timerTaskExecutor" class="org.springframework.scheduling.timer.TimerTaskExecutor">
<property name="delay" value="10000" />
</bean>

在集成Quartz任务调度框架过程中，Spring引入了SimpleThreadPoolTaskExecutor。此时，用户提交的所有任务将交由Quartz维护的线程池执行，其示例配置如下。很显然，开发者能够控制这一线程池的具体细节。在使用Quartz实现任务调度过程中，它会使用内置的线程池处理任务。

<bean id="simpleThreadPoolTaskExecutor" class="org.springframework.scheduling.quartz.SimpleThreadPoolTaskExecutor">
<property name="makeThreadsDaemons" value="true"/>
<property name="threadCount" value="5" />
<property name="threadNamePrefix" value="simpleThreadPoolTaskExecutor"/>
<property name="waitForJobsToCompleteOnShutdown" value="true" />
</bean>

如果开发者在使用BEA WebLogic 9+和IBM WebSphere 6+版本的Java EE应用服务器，则应用通过JNDI树还能够获得CommonJ提供的Java EE容器受管连接池。我们之前使用的TimerTaskExecutor、SimpleThreadPoolTaskExecutor并没有使用到容器受管连接池，而且它们也不能够使用到Java EE应用的上下文信息，而CommonJ却可以做到。甚至，CommonJ同时支持同步和异步处理用户提交的任务。当然，CommonJ并不是Java EE规范的组成部分，因此只有BEA和IBM客户才能够享受到这一待遇，即WorkManagerTaskExecutor。

WorkManagerTaskExecutor的示例配置如下。我们可以在很多场合使用到这一容器受管的线程池服务，比如，当使用Spring JMS集成开发消息应用时，开发者可以在WebSphere或WebLogic中使用WorkManagerTaskExecutor提供的线程池服务。

<bean id="workManagerTaskExecutor" class="org.springframework.scheduling.commonj.WorkManagerTaskExecutor">
<property name="workManagerName" value="jndi/workManager" />
</bean>

在涉及到任务调度的应用场合，必然会使用到线程池服务。开发者在使用TaskExecutor抽象过程中，它们只需要借助于TaskExecutor提交任务，而任务的调度、执行根本不用开发者操心。由于上述提供线程池服务的各种TaskExecutor都实现了DisposableBean回调接口，因此一旦Java EE容器关闭，相应的线程池也会被优雅地终结。在介绍了Spring提供的TaskExecutor抽象后，我们再来研究Spring提供的任务调度支持。Spring 2.0本身并没有实现任务调度服务，它只是对现有的各种任务调度技术、框架进行了集成、封装，从而简化它们的使用，并采用DI配置形式驱动它们。由于Spring imagedb示例应用展示了任务调度支持，因此我们在这里将结合Eclipse imagedb项目展开论述。

public void saveList(JshxPhoneInterface jshxPhoneInterface){
		list.add(jshxPhoneInterface);
		saveLogs();
	}
	
	private void saveLogs(){
		TaskExecutor threadPool = SpringContextHolder.getBean("threadPoolTaskExecutor");
		threadPool.execute(new Runnable(){
			public void run() {
				if (list.size() >= count){
					try {
						JshxPhoneInterfaceService jshxPhoneInterfaceService = (JshxPhoneInterfaceService) SpringContextHolder.getBean("jshxPhoneInterfaceService");
						jshxPhoneInterfaceService.saveAll(list);
					} catch (Exception e) {
						e.printStackTrace();
					}
					list.clear();
				}
			}
		});
	}