java-blog

 昨天開始研究java的多線程包java.util.concurrent，根據(jù)自己的理解實(shí)現(xiàn)了一個(gè)消息隊(duì)列異步調(diào)用(200行代碼左右)。拿出來與大家分享我的勞動成果。
  希望大家多提意見。指出哪里寫的不好，以后加以改正。

ThreadPoolManager類：負(fù)責(zé)管理線程池，調(diào)用輪詢的線程來訪問字符串緩沖區(qū)的內(nèi)容，維護(hù)緩沖區(qū)，當(dāng)線程池溢出時(shí)拋出的Runnable任務(wù)被加入到字符緩沖區(qū)。
public class ThreadPoolManager
{
    private static ThreadPoolManager tpm = new ThreadPoolManager();

    // 線程池維護(hù)線程的最少數(shù)量
    private final static int CORE_POOL_SIZE = 4;

    // 線程池維護(hù)線程的最大數(shù)量
    private final static int MAX_POOL_SIZE = 10;

    // 線程池維護(hù)線程所允許的空閑時(shí)間
    private final static int KEEP_ALIVE_TIME = 0;

    // 線程池所使用的緩沖隊(duì)列大小
    private final static int WORK_QUEUE_SIZE = 10;

    // 消息緩沖隊(duì)列
    Queue<String> msgQueue = new LinkedList<String>();

    // 訪問消息緩存的調(diào)度線程
    final Runnable accessBufferThread = new Runnable()
    {
        public void run()
        {
            // 查看是否有待定請求，如果有，則創(chuàng)建一個(gè)新的AccessDBThread，并添加到線程池中
            if( hasMoreAcquire() )
            {
                String msg = ( String ) msgQueue.poll();
                Runnable task = new AccessDBThread( msg );
                threadPool.execute( task );
            }
        }
    };

    final RejectedExecutionHandler handler = new RejectedExecutionHandler()
    {
        public void rejectedExecution( Runnable r, ThreadPoolExecutor executor )
        {
            System.out.println(((AccessDBThread )r).getMsg()+"消息放入隊(duì)列中重新等待執(zhí)行");
            msgQueue.offer((( AccessDBThread ) r ).getMsg() );
        }
    };

    // 管理數(shù)據(jù)庫訪問的線程池
    final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
            CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
            new ArrayBlockingQueue( WORK_QUEUE_SIZE ), this.handler );

    // 調(diào)度線程池
    final ScheduledExecutorService scheduler = Executors
            .newScheduledThreadPool( 1 );

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

    public static ThreadPoolManager newInstance()
    {
        return tpm;
    }

    private ThreadPoolManager(){}

    private boolean hasMoreAcquire()
    {
        return !msgQueue.isEmpty();
    }

    public void addLogMsg( String msg )
    {
        Runnable task = new AccessDBThread( msg );
        threadPool.execute( task );
    }
}

public class AccessDBThread implements Runnable
{
    private String msg;
    
    public String getMsg()
    {
        return msg;
    }

    public void setMsg( String msg )
    {
        this.msg = msg;
    }
    
    public AccessDBThread(){
        super();
    }
    
    public AccessDBThread(String msg){
        this.msg = msg;
    }

    public void run()
    {
        // 向數(shù)據(jù)庫中添加Msg變量值
        System.out.println("Added the message: "+msg+" into the Database");
    }

}

public class TestDriver
{
    ThreadPoolManager tpm = ThreadPoolManager.newInstance();

    public void sendMsg( String msg )
    {
        tpm.addLogMsg( msg + "記錄一條日志 " );
    }

    public static void main( String[] args )
    {
        for( int i = 0; i < 100; i++ )
        {
            new TestDriver().sendMsg( Integer.toString( i ) );
        }
    }
}