Java实现通用线程池

线程池通俗的描述就是预先创建若干空闲线程 等到需要用多线程去处理事务的时候去唤醒某些空闲线程执行处理任务 这样就省去了频繁创建线程的时间 因为频 繁创建线程是要耗费大量的CPU资源的 如果一个应用程序需要频繁地处理大量并发事务 不断的创建销毁线程往往会大大地降低系统的效率 这时候线程池就派 上用场了

本文旨在使用Java语言编写一个通用的线程池 当需要使用线程池处理事务时 只需按照指定规范封装好事务处理对象 然后用已有的线程池对象去自动选择空 闲线程自动调用事务处理对象即可 并实现线程池的动态修改（修改当前线程数 最大线程数等） 下面是实现代码

//ThreadTask java

package polarman threadpool

/** *//**

*线程任务

* @author ryang

*

*/

public interface ThreadTask {

public void run()

}

//PooledThread java

package polarman threadpool

import java util Collectionimport java util Vector

/** *//**

*接受线程池管理的线程

* @author ryang

*

*/

public class PooledThread extends Thread {

protected Vector tasks = new Vector()

protected boolean running = false

protected boolean stopped = false

protected boolean paused = false

protected boolean killed = false

private ThreadPool pool

public PooledThread(ThreadPool pool) { this pool = pool

}

public void putTask(ThreadTask task) { tasks add(task)

}

public void putTasks(ThreadTask[] tasks) { for(int i= i<tasks lengthi++) this tasks add(tasks[i])

}

public void putTasks(Collection tasks) { this tasks addAll(tasks)

}

protected ThreadTask popTask() { if(tasks size() >) return (ThreadTask)tasks remove( )

else

return null

}

public boolean isRunning() {

return running

}

public void stopTasks() {

stopped = true

}

public void stopTasksSync() {

stopTasks()

while(isRunning()) { try {

sleep( )

} catch (InterruptedException e) {

}

}

}

public void pauseTasks() {

paused = true

}

public void pauseTasksSync() {

pauseTasks()

while(isRunning()) { try {

sleep( )

} catch (InterruptedException e) {

}

}

}

public void kill() { if(!running)

interrupt()

else

killed = true

}

public void killSync() {

kill()

while(isAlive()) { try {

sleep( )

} catch (InterruptedException e) {

}

}

}

public synchronized void startTasks() {

running = true

this notify()

}

public synchronized void run() { try { while(true) { if(!running || tasks size() == ) { pool notifyForIdleThread()//System out println(Thread currentThread() getId() + : 空闲 )this wait()}else {

ThreadTask task

while((task = popTask()) != null) { task run()if(stopped) {

stopped = false

if(tasks size() >) { tasks clear()System out println(Thread currentThread() getId() + : Tasks are stopped )

break

}

}

if(paused) {

paused = false

if(tasks size() >) { System out println(Thread currentThread() getId() + : Tasks are paused )

break

}

}

}

running = false

}

if(killed) {

killed = false

break

}

}

}catch(InterruptedException e) {

return

}

//System out println(Thread currentThread() getId() + : Killed )

}

}

//ThreadPool java

package polarman threadpool

import java util Collectionimport java util Iteratorimport java util Vector

/** *//**

*线程池

* @author ryang

*

*/

public class ThreadPool {

protected int maxPoolSize

protected int initPoolSize

protected Vector threads = new Vector()

protected boolean initialized = false

protected boolean hasIdleThread = false

public ThreadPool(int maxPoolSize int initPoolSize) { this maxPoolSize = maxPoolSizethis initPoolSize = initPoolSize

}

public void init() {

initialized = true

for(int i= i<initPoolSizei++) {

PooledThread thread = new PooledThread(this)

thread start()threads add(thread)

}

//System out println( 线程池初始化结束 线程数= + threads size() + 最大线程数= + maxPoolSize)

}

public void setMaxPoolSize(int maxPoolSize) { //System out println( 重设最大线程数 最大线程数= + maxPoolSize)this maxPoolSize = maxPoolSize

if(maxPoolSize <getPoolSize())

setPoolSize(maxPoolSize)

}

/** *//**

*重设当前线程数

* 若需杀掉某线程 线程不会立刻杀掉 而会等到线程中的事务处理完成* 但此方法会立刻从线程池中移除该线程 不会等待事务处理结束

* @param size

*/

public void setPoolSize(int size) { if(!initialized) {

initPoolSize = size

return

}else if(size >getPoolSize()) { for(int i=getPoolSize()i<size &&i<maxPoolSizei++) {

PooledThread thread = new PooledThread(this)

thread start()threads add(thread)

}

}else if(size <getPoolSize()) { while(getPoolSize() >size) { PooledThread th = (PooledThread)threads remove( )th kill()

}

}

//System out println( 重设线程数 线程数= + threads size())

}

public int getPoolSize() { return threads size()

}

protected void notifyForIdleThread() {

hasIdleThread = true

}

protected boolean waitForIdleThread() {

hasIdleThread = false

while(!hasIdleThread &&getPoolSize() >= maxPoolSize) { try { Thread sleep( )} catch (InterruptedException e) {

return false

}

}

return true

}

public synchronized PooledThread getIdleThread() { while(true) { for(Iterator itr=erator()itr hasNext()) { PooledThread th = (PooledThread)itr next()if(!th isRunning())

return th

}

if(getPoolSize() <maxPoolSize) {

PooledThread thread = new PooledThread(this)

thread start()threads add(thread)

return thread

}

//System out println( 线程池已满 等待 )

if(waitForIdleThread() == false)

return null

}

}

public void processTask(ThreadTask task) {

PooledThread th = getIdleThread()

if(th != null) { th putTask(task)th startTasks()

}

}

public void processTasksInSingleThread(ThreadTask[] tasks) {

PooledThread th = getIdleThread()

if(th != null) { th putTasks(tasks)th startTasks()

}

}

public void processTasksInSingleThread(Collection tasks) {

PooledThread th = getIdleThread()

if(th != null) { th putTasks(tasks)th startTasks()

}

}

}

下面是线程池的测试程序

//ThreadPoolTest java

import java io BufferedReaderimport java io IOExceptionimport java io InputStreamReader

import polarman threadpool ThreadPoolimport polarman threadpool ThreadTask

public class ThreadPoolTest {

public static void main(String[] args) { System out println( quit 退出 )System out println( task A 启动任务A 时长为 秒 )System out println( size 设置当前线程池大小为 )System out println( max 设置线程池最大线程数为 )System out println()

final ThreadPool pool = new ThreadPool( )pool init()

Thread cmdThread = new Thread() { public void run() {

BufferedReader reader = new BufferedReader(new InputStreamReader(System in))

while(true) { try { String line = reader readLine()String words[] = line split( )if(words[ ] equalsIgnoreCase( quit )) { System exit( )}else if(words[ ] equalsIgnoreCase( size ) &&words length >= ) { try { int size = Integer parseInt(words[ ])pool setPoolSize(size)}catch(Exception e) {

}

}else if(words[ ] equalsIgnoreCase( max ) &&words length >= ) { try { int max = Integer parseInt(words[ ])pool setMaxPoolSize(max)}catch(Exception e) {

}

}else if(words[ ] equalsIgnoreCase( task ) &&words length >= ) { try { int timelen = Integer parseInt(words[ ])SimpleTask task = new SimpleTask(words[ ] timelen * )pool processTask(task)}catch(Exception e) {

}

}

} catch (IOException e) { e printStackTrace()

}

}

}

}

cmdThread start()

/**//*

for(int i= i<i++){

SimpleTask task = new SimpleTask( Task + i (i+ )* )pool processTask(task)

}*/

}

}

class SimpleTask implements ThreadTask {

private String taskName

private int timeLen

public SimpleTask(String taskName int timeLen) { this taskName = taskNamethis timeLen = timeLen

}

public void run() { System out println(Thread currentThread() getId() +

: START TASK + taskName + )

try { Thread sleep(timeLen)} catch (InterruptedException e) {

}

System out println(Thread currentThread() getId() +

: END TASK + taskName + )

}

}

使用此线程池相当简单 下面两行代码初始化线程池

ThreadPool pool = new ThreadPool( )pool init()

要处理的任务实现ThreadTask 接口即可（如测试代码里的SimpleTask） 这个接口只有一个方法run()

两行代码即可调用

lishixinzhi/Article/program/Java/hx/201311/27203

1、基本概念及原理

由上面的分析可以看出，问题的根源就在于对数据库连接资源的低效管理。我们知道，

对于共享资源，有一个很著名的设计模式：资源池（Resource Pool）。该模式正是为了解决资源的频繁分配?释放所造成的问题。为解决上述问题，可以采用数据库连接池技术。数据库连接池的基本思想就是为数据库连接建立一个“缓冲池”。预先在缓冲池中放入一定数量的连接，当需要建立数据库连接时，只需从“缓冲池”中取出一个，使用完毕之后再放回去。我们可以通过设定连接池最大连接数来防止系统无尽的与数据库连接。更为重要的是我们可以通过连接池的管理机制监视数据库的连接的数量?使用情况，为系统开发?测试及性能调整提供依据。

连接池的基本工作原理

2、服务器自带的连接池

JDBC的API中没有提供连接池的方法。一些大型的WEB应用服务器如BEA的WebLogic和IBM的WebSphere等提供了连接池的机制，但是必须有其第三方的专用类方法支持连接池的用法。

连接池关键问题分析

1、并发问题

为了使连接管理服务具有最大的通用性，必须考虑多线程环境，即并发问题。这个问题相对比较好解决，因为Java语言自身提供了对并发管理的支持，使用 synchronized关键字即可确保线程是同步的。使用方法为直接在类方法前面加上synchronized关键字，如：

public synchronized Connection getConnection（）

2、多数据库服务器和多用户

对于大型的企业级应用，常常需要同时连接不同的数据库（如连接Oracle和Sybase）。如何连接不同的数据库呢？我们采用的策略是：设计一个符合单例模式的连接池管理类，在连接池管理类的唯一实例被创建时读取一个资源文件，其中资源文件中存放着多个数据库的url地址（）?用户名（）?密码（）等信息。如tx.url=172.21.15.123：5000/tx_it，tx.user=yang，tx.password=yang321。根据资源文件提供的信息，创建多个连接池类的实例，每一个实例都是一个特定数据库的连接池。连接池管理类实例为每个连接池实例取一个名字，通过不同的名字来管理不同的连接池。

对于同一个数据库有多个用户使用不同的名称和密码访问的情况，也可以通过资源文件处理，即在资源文件中设置多个具有相同url地址，但具有不同用户名和密码的数据库连接信息。

3、事务处理

我们知道，事务具有原子性，此时要求对数据库的 *** 作符合“ALL-ALL-NOTHING”原则,即对于一组SQL语句要么全做，要么全不做。

在Java语言中，Connection类本身提供了对事务的支持，可以通过设置Connection的AutoCommit属性为false,然后显式的调用commit或rollback方法来实现。但要高效的进行Connection复用，就必须提供相应的事务支持机制。可采用每一个事务独占一个连接来实现，这种方法可以大大降低事务管理的复杂性。

4、连接池的分配与释放

连接池的分配与释放，对系统的性能有很大的影响。合理的分配与释放，可以提高连接的复用度，从而降低建立新连接的开销，同时还可以加快用户的访问速度。

对于连接的管理可使用空闲池。即把已经创建但尚未分配出去的连接按创建时间存放到一个空闲池中。每当用户请求一个连接时，系统首先检查空闲池内有没有空闲连接。如果有就把建立时间最长（通过容器的顺序存放实现）的那个连接分配给他（实际是先做连接是否有效的判断，如果可用就分配给用户，如不可用就把这个连接从空闲池删掉，重新检测空闲池是否还有连接）；如果没有则检查当前所开连接池是否达到连接池所允许的最大连接数（maxConn）,如果没有达到，就新建一个连接，如果已经达到，就等待一定的时间（timeout）。如果在等待的时间内有连接被释放出来就可以把这个连接分配给等待的用户，如果等待时间超过预定时间timeout,则返回空值（null）。系统对已经分配出去正在使用的连接只做计数，当使用完后再返还给空闲池。对于空闲连接的状态，可开辟专门的线程定时检测，这样会花费一定的系统开销，但可以保证较快的响应速度。也可采取不开辟专门线程，只是在分配前检测的方法。

5、连接池的配置与维护

连接池中到底应该放置多少连接，才能使系统的性能最佳？系统可采取设置最小连接数（minConn）和最大连接数（maxConn）来控制连接池中的连接。最小连接数是系统启动时连接池所创建的连接数。如果创建过多，则系统启动就慢，但创建后系统的响应速度会很快；如果创建过少，则系统启动的很快，响应起来却慢。这样，可以在开发时，设置较小的最小连接数，开发起来会快，而在系统实际使用时设置较大的，因为这样对访问客户来说速度会快些。最大连接数是连接池中允许连接的最大数目，具体设置多少，要看系统的访问量，可通过反复测试，找到最佳点。

如何确保连接池中的最小连接数呢？有动态和静态两种策略。动态即每隔一定时间就对连接池进行检测，如果发现连接数量小于最小连接数，则补充相应数量的新连接,以保证连接池的正常运转。静态是发现空闲连接不够时再去检查。

连接池的实现

1、连接池模型

本文讨论的连接池包括一个连接池类（DBConnectionPool）和一个连接池管理类（DBConnetionPoolManager）。连接池类是对某一数据库所有连接的“缓冲池”，主要实现以下功能：①从连接池获取或创建可用连接；②使用完毕之后，把连接返还给连接池；③在系统关闭前，断开所有连接并释放连接占用的系统资源；④还能够处理无效连接（原来登记为可用的连接，由于某种原因不再可用，如超时，通讯问题），并能够限制连接池中的连接总数不低于某个预定值和不超过某个预定值。

连接池管理类是连接池类的外覆类（wrapper）,符合单例模式，即系统中只能有一个连接池管理类的实例。其主要用于对多个连接池对象的管理，具有以下功能：①装载并注册特定数据库的JDBC驱动程序；②根据属性文件给定的信息，创建连接池对象；③ 为方便管理多个连接池对象，为每一个连接池对象取一个名字，实现连接池名字与其实例之间的映射；④跟踪客户使用连接情况，以便需要是关闭连接释放资源。连接池管理类的引入主要是为了方便对多个连接池的使用和管理，如系统需要连接不同的数据库，或连接相同的数据库但由于安全性问题，需要不同的用户使用不同的名称和密码。

配置包括3步：1.让tomcat容器启动创建数据库连接池2.在某个项目中关联数据库连接池，3.取得数据库连接池并使用。使用包括，当拿到数据库连接后，可以通过2种方式来使用，1.使用jstl的标签，2.封装成返回connection的方法。

1.让tomcat容器启动创建数据库连接池：

配置的地方很多，根据不同的需求配置在不同的位置，但原理就一个，就是要配置一个Context标签下有一个resource标签。

a.在Tomcat 6.0\conf\Catalina\localhost目录下新建一个跟项目名一样的*.xml文件

b.数据源设置的名字(JNDI) auth:表示数据源由谁管理，这儿是tomcat容器本身 type：类型

c.驱动设置，该驱动放在tomcat的lib下或者放在该站点的lib下设置连接数据库字符串，url设置连接数据库的用户名设置数据库连接的密码-->

2.在某个项目中关联数据库连接池，在web.xml添加代码：

<!--关联jndi数据库连接池-->

3.使用jstl标签测试数据源是否配置成功

4.在jsp中通过java代码测试：

5.封装成bean对外提供得到数据库连接池连接的方式：

---