（转）使用java.util.concurrent实现的线程池、消息队列功能

（转）使用java.util.concurrent实现的线程池、消息队列功能

昨天开始研究java.util.concurrent，是出于线程安全的知识懂得不多，对自己写的线程池没有信心，所以就用了包里专家写好的线程池。这个包的功能很强大。有兴趣的朋友可以搜索了解更多的内容。
     今天刚写好了一段200行左右的代码，拿出来跟大家分享我的学习经验。初次实践，不足之处，望能得到高手的指点。
功能说明：一个发送消息模块将消息发送到消息队列中，无需等待返回结果，发送模块继续执行其他任务。消息队列中的指令由线程池中的线程来处理。使用一个Queue来存放线程池溢出时的任务。
TestDriver.java是一个驱动测试，sendMsg方法不间断的向ThreadPoolManager发送数据。
01
public class TestDriver
02
{
03
    ThreadPoolManager tpm = ThreadPoolManager.newInstance();
04

05
    public void sendMsg( String msg )
06
    {
07
        tpm.addLogMsg( msg + "记录一条日志 " );
08
    }
09

10
    public static void main( String[] args )
11
    {
12
        for( int i = 0; i < 100; i++ )
13
        {
14
            new TestDriver().sendMsg( Integer.toString( i ) );
15
        }
16
    }
17
}


ThreadPoolManager类：是负责管理线程池的类。同时维护一个Queue和调度进程。

01
public class ThreadPoolManager
02
{
03
private static ThreadPoolManager tpm = new ThreadPoolManager();
04

05
// 线程池维护线程的最少数量
06
private final static int CORE_POOL_SIZE = 4;
07

08
// 线程池维护线程的最大数量
09
private final static int MAX_POOL_SIZE = 10;
10

11
// 线程池维护线程所允许的空闲时间
12
private final static int KEEP_ALIVE_TIME = 0;
13

14
// 线程池所使用的缓冲队列大小
15
private final static int WORK_QUEUE_SIZE = 10;
16

17
// 消息缓冲队列
18
Queue<String> msgQueue = new LinkedList<String>();
19

20
// 访问消息缓存的调度线程
21
final Runnable accessBufferThread = new Runnable()
22
{
23
  public void run()
24
  {
25
   // 查看是否有待定请求，如果有，则创建一个新的AccessDBThread，并添加到线程池中
26
   if( hasMoreAcquire() )
27
   {
28
    String msg = ( String ) msgQueue.poll();
29
    Runnable task = new AccessDBThread( msg );
30
    threadPool.execute( task );
31
   }
32
  }
33
};
34

35
final RejectedExecutionHandler handler = new RejectedExecutionHandler()
36
{
37
  public void rejectedExecution( Runnable r, ThreadPoolExecutor executor )
38
  {
39
   System.out.println(((AccessDBThread )r).getMsg()+"消息放入队列中重新等待执行");
40
   msgQueue.offer((( AccessDBThread ) r ).getMsg() );
41
  }
42
};
43

44
// 管理数据库访问的线程池
45
final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
46
   CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
47
   new ArrayBlockingQueue( WORK_QUEUE_SIZE ), this.handler );
48

49
// 调度线程池
50
final ScheduledExecutorService scheduler = Executors
51
   .newScheduledThreadPool( 1 );
52

53
final ScheduledFuture taskHandler = scheduler.scheduleAtFixedRate(
54
   accessBufferThread, 0, 1, TimeUnit.SECONDS );
55

56
public static ThreadPoolManager newInstance()
57
{
58
  return tpm;
59
}
60

61
private ThreadPoolManager(){}
62

63
private boolean hasMoreAcquire()
64
{
65
  return !msgQueue.isEmpty();
66
}
67

68
public void addLogMsg( String msg )
69
{
70
  Runnable task = new AccessDBThread( msg );
71
  threadPool.execute( task );
72
}
73
}


AccessDBThread类：线程池中工作的线程。

01
public class AccessDBThread implements Runnable
02
{
03
private String msg;
04
 
05
public String getMsg()
06
{
07
  return msg;
08
}
09

10
public void setMsg( String msg )
11
{
12
  this.msg = msg;
13
}
14
 
15
public AccessDBThread(){
16
  super();
17
}
18
 
19
public AccessDBThread(String msg){
20
  this.msg = msg;
21
}
22

23
public void run()
24
{
25
  // 向数据库中添加Msg变量值
26
  System.out.println("Added the message: "+msg+" into the Database");
27
}
28

29
}