( )根據xml文件來管理線程池的最大最小線程數( )對線程池通過Timer定期掃描以防止線程未激活 ( )通過某一個變量(本程序中是freeThreadCount)來得到空閑線程的數目 一 配置xml(listen xml)是 ?xml version= encoding= UTF ?configConsumeThreadPoolminPools /minPools ! 線程池最小線程 maxPools /maxPools! 線程池最大線程 checkThreadPeriod /checkThreadPeriod ! 檢查線程池中線程的周期 分鐘 /ConsumeThreadPool/config 二 對于ConsumeThreadPoolPara的javabean: import java io *;public class ConsumeThreadPoolPara implements Serializable{private int minPools;private int maxPools;private int checkThreadPeriod;public int getMinPools(){return minPools;}public int getMaxPools(){return maxPools;}public int getCheckThreadPeriod(){return checkThreadPeriod;}public void setMinPools(int minPools){this minPools = minPools;}public void setMaxPools(int maxPools){this maxPools = maxPools;}public void setCheckThreadPeriod(int checkThreadPeriod){this checkThreadPeriod = checkThreadPeriod;}public String toString(){return minPools+ + maxPools+ +checkThreadPeriod;}public ConsumeThreadPoolPara() {}public static void main(String[] args) {ConsumeThreadPoolPara consumeThreadPool = new ConsumeThreadPoolPara();}} 三 解析xml程序代碼(生成ConsumeThreadPoolPara) 使用jdom解析 import jdom *;import jdom input SAXBuilder;import java io *;import java util *;public class ParseConfig {static Hashtable Listens = null;static ConnPara connpara = null;static ConsumeThreadPoolPara consumeThreadPoolPara = null;private static String configxml = listen xml ;static{getConsumeThreadPoolPara(); //得到消費的線程池的參數}/*** 裝載文檔* @return 返回根結點* @throws JDOMException*/public static Element loadDocument() throws JDOMException{SAXBuilder parser = new SAXBuilder(); // 新建立構造器try {Document document = parser build(configxml);Element root = document getRootElement();return root;}catch(JDOMException e){logger error( listen xml文件格式非法! );throw new JDOMException();}}public static ConsumeThreadPoolPara getConsumeThreadPoolPara(){if(consumeThreadPoolPara ==null){try {Element root = loadDocument();Element consumeThreadPool = root getChild( ConsumeThreadPool );if (consumeThreadPool != null) { //代表有數據庫配置consumeThreadPoolPara = new ConsumeThreadPoolPara();Element minPools = consumeThreadPool getChild( minPools );consumeThreadPoolPara setMinPools(Integer parseInt(minPools getTextTrim()));Element maxPools = consumeThreadPool getChild( maxPools );consumeThreadPoolPara setMaxPools(Integer parseInt(maxPools getTextTrim()));Element checkThreadPeriod = consumeThreadPool getChild( checkThreadPeriod );consumeThreadPoolPara setCheckThreadPeriod(Integer parseInt(checkThreadPeriod getTextTrim()));}}catch (JDOMException e) {}}return consumeThreadPoolPara;}} 四 線程池源代碼 import java util *;/*** pTitle: 線程池/p* pDescription: 采集消費模塊/p* pCopyright: Copyright (c) /p* pCompany: /p* @author 張榮斌* @version */public class ThreadPool {private static int minPools = ; //最小連接池數目private static int maxPools = ; //最大連接池數目private static int checkThreadPeriod = ; //檢查連接池的周期ArrayList m_ThreadList; //工作線程列表LinkedList m_RunList = null; //工作任務列表int totalThread = ; //總線程數static int freeThreadCount = ; //未被使用的線程數目private java util Timer timer = null; //定時器static Object o = new Object();static{ //先初始化線程池的參數ConsumeThreadPoolPara consumeThreadPoolPara = ParseConfig getConsumeThreadPoolPara();if(consumeThreadPoolPara!=null){minPools = consumeThreadPoolPara getMinPools();maxPools = consumeThreadPoolPara getMaxPools();checkThreadPeriod = consumeThreadPoolPara getCheckThreadPeriod()* * ;}}public void setMinPools(int minPools){this minPools = minPools;}public void setMaxPools(int maxPools){this maxPools = maxPools;}public void setCheckThreadPeriod(int checkThreadPeriod){this checkThreadPeriod = checkThreadPeriod;}public ThreadPool() {m_ThreadList=new ArrayList();m_RunList=new LinkedList();for(int i= ;iminPools;i++){WorkerThread temp=new WorkerThread();totalThread = totalThread + ;m_ThreadList add(temp);temp start();try{Thread sleep( );}catch(Exception e){}}timer = new Timer(true); //啟動定時器timer schedule(new CheckThreadTask(this) checkThreadPeriod);}/*** 當有一個工作來的時候啟動線程池的線程* 當空閑線程數為 的時候 看總線程是否小于最大線程池的數目 就new一個新的線程 否則sleep 直到有空閑線程為止;* 當空閑線程不為 則將任務丟給空閑線程去完成* @param work*/public synchronized void run(String work){if (freeThreadCount == ) {if(totalThreadmaxPools){WorkerThread temp = new WorkerThread();totalThread = totalThread + ;m_ThreadList add(temp);temp start();synchronized(m_RunList){m_RunList add(work);m_RunList notify();}}else{while (freeThreadCount == ) {try {Thread sleep( );}catch (InterruptedException e) {}}synchronized(m_RunList){m_RunList add(work);m_RunList notify();}}} else {synchronized(m_RunList){m_RunList add(work);m_RunList notify();}}}/*** 檢查所有的線程的有效性*/public synchronized void checkAllThreads() {Iterator lThreadIterator = erator();while (lThreadIterator hasNext()) { //逐個遍厲WorkerThread lTestThread = (WorkerThread) lThreadIterator next();if (! (lTestThread isAlive())) { //如果處在非活動狀態時lTestThread = new WorkerThread(); //重新生成個線程lTestThread start(); //啟動}}}/*** 打印調試信息*/public void printDebugInfo(){System out println( totalThread= +totalThread);System out println( m_ThreadList size()= +m_ThreadList size());}/**** pTitle: 工作線程類/p* @author 張榮斌* @version */class WorkerThread extends Thread{boolean running = true;String work;public void run(){while(running){synchronized(o){freeThreadCount++;}synchronized(m_RunList){while(m_RunList size() == ){try{m_RunList wait();if(!running) return;}catch(InterruptedException e){} lishixinzhi/Article/program/Java/gj/201311/27379
成都創新互聯公司服務項目包括天涯網站建設、天涯網站制作、天涯網頁制作以及天涯網絡營銷策劃等。多年來,我們專注于互聯網行業,利用自身積累的技術優勢、行業經驗、深度合作伙伴關系等,向廣大中小型企業、政府機構等提供互聯網行業的解決方案,天涯網站推廣取得了明顯的社會效益與經濟效益。目前,我們服務的客戶以成都為中心已經輻射到天涯省份的部分城市,未來相信會繼續擴大服務區域并繼續獲得客戶的支持與信任!
線程池簡介:
多線程技術主要解決處理器單元內多個線程執行的問題,它可以顯著減少處理器單元的閑置時間,增加處理器單元的吞吐能力。
假設一個服務器完成一項任務所需時間為:T1 創建線程時間,T2 在線程中執行任務的時間,T3 銷毀線程時間。
如果:T1 + T3 遠大于 T2,則可以采用線程池,以提高服務器性能。
一個線程池包括以下四個基本組成部分:
1、線程池管理器(ThreadPool):用于創建并管理線程池,包括 創建線程池,銷毀線程池,添加新任務;
2、工作線程(PoolWorker):線程池中線程,在沒有任務時處于等待狀態,可以循環的執行任務;
3、任務接口(Task):每個任務必須實現的接口,以供工作線程調度任務的執行,它主要規定了任務的入口,任務執行完后的收尾工作,任務的執行狀態等;
4、任務隊列(taskQueue):用于存放沒有處理的任務。提供一種緩沖機制。
線程池技術正是關注如何縮短或調整T1,T3時間的技術,從而提高服務器程序性能的。它把T1,T3分別安排在服務器程序的啟動和結束的時間段或者一些空閑的時間段,這樣在服務器程序處理客戶請求時,不會有T1,T3的開銷了。
線程池不僅調整T1,T3產生的時間段,而且它還顯著減少了創建線程的數目,看一個例子:
假設一個服務器一天要處理50000個請求,并且每個請求需要一個單獨的線程完成。在線程池中,線程數一般是固定的,所以產生線程總數不會超過線程池中線程的數目,而如果服務器不利用線程池來處理這些請求則線程總數為50000。一般線程池大小是遠小于50000。所以利用線程池的服務器程序不會為了創建50000而在處理請求時浪費時間,從而提高效率。
代碼實現中并沒有實現任務接口,而是把Runnable對象加入到線程池管理器(ThreadPool),然后剩下的事情就由線程池管理器(ThreadPool)來完成了
package?mine.util.thread;??
import?java.util.LinkedList;??
import?java.util.List;??
/**?
*?線程池類,線程管理器:創建線程,執行任務,銷毀線程,獲取線程基本信息?
*/??
public?final?class?ThreadPool?{??
//?線程池中默認線程的個數為5??
private?static?int?worker_num?=?5;??
//?工作線程??
private?WorkThread[]?workThrads;??
//?未處理的任務??
private?static?volatile?int?finished_task?=?0;??
//?任務隊列,作為一個緩沖,List線程不安全??
private?ListRunnable?taskQueue?=?new?LinkedListRunnable();??
private?static?ThreadPool?threadPool;??
//?創建具有默認線程個數的線程池??
private?ThreadPool()?{??
this(5);??
}??
//?創建線程池,worker_num為線程池中工作線程的個數??
private?ThreadPool(int?worker_num)?{??
ThreadPool.worker_num?=?worker_num;??
workThrads?=?new?WorkThread[worker_num];??
for?(int?i?=?0;?i??worker_num;?i++)?{??
workThrads[i]?=?new?WorkThread();??
workThrads[i].start();//?開啟線程池中的線程??
}??
}??
//?單態模式,獲得一個默認線程個數的線程池??
public?static?ThreadPool?getThreadPool()?{??
return?getThreadPool(ThreadPool.worker_num);??
}??
//?單態模式,獲得一個指定線程個數的線程池,worker_num(0)為線程池中工作線程的個數??
//?worker_num=0創建默認的工作線程個數??
public?static?ThreadPool?getThreadPool(int?worker_num1)?{??
if?(worker_num1?=?0)??
worker_num1?=?ThreadPool.worker_num;??
if?(threadPool?==?null)??
threadPool?=?new?ThreadPool(worker_num1);??
return?threadPool;??
}??
//?執行任務,其實只是把任務加入任務隊列,什么時候執行有線程池管理器覺定??
public?void?execute(Runnable?task)?{??
synchronized?(taskQueue)?{??
taskQueue.add(task);??
taskQueue.notify();??
}??
}??
//?批量執行任務,其實只是把任務加入任務隊列,什么時候執行有線程池管理器覺定??
public?void?execute(Runnable[]?task)?{??
synchronized?(taskQueue)?{??
for?(Runnable?t?:?task)??
taskQueue.add(t);??
taskQueue.notify();??
}??
}??
//?批量執行任務,其實只是把任務加入任務隊列,什么時候執行有線程池管理器覺定??
public?void?execute(ListRunnable?task)?{??
synchronized?(taskQueue)?{??
for?(Runnable?t?:?task)??
taskQueue.add(t);??
taskQueue.notify();??
}??
}??
//?銷毀線程池,該方法保證在所有任務都完成的情況下才銷毀所有線程,否則等待任務完成才銷毀??
public?void?destroy()?{??
while?(!taskQueue.isEmpty())?{//?如果還有任務沒執行完成,就先睡會吧??
try?{??
Thread.sleep(10);??
}?catch?(InterruptedException?e)?{??
e.printStackTrace();??
}??
}??
//?工作線程停止工作,且置為null??
for?(int?i?=?0;?i??worker_num;?i++)?{??
workThrads[i].stopWorker();??
workThrads[i]?=?null;??
}??
threadPool=null;??
taskQueue.clear();//?清空任務隊列??
}??
//?返回工作線程的個數??
public?int?getWorkThreadNumber()?{??
return?worker_num;??
}??
//?返回已完成任務的個數,這里的已完成是只出了任務隊列的任務個數,可能該任務并沒有實際執行完成??
public?int?getFinishedTasknumber()?{??
return?finished_task;??
}??
//?返回任務隊列的長度,即還沒處理的任務個數??
public?int?getWaitTasknumber()?{??
return?taskQueue.size();??
}??
//?覆蓋toString方法,返回線程池信息:工作線程個數和已完成任務個數??
@Override??
public?String?toString()?{??
return?"WorkThread?number:"?+?worker_num?+?"??finished?task?number:"??
+?finished_task?+?"??wait?task?number:"?+?getWaitTasknumber();??
}??
/**?
*?內部類,工作線程?
*/??
private?class?WorkThread?extends?Thread?{??
//?該工作線程是否有效,用于結束該工作線程??
private?boolean?isRunning?=?true;??
/*?
*?關鍵所在啊,如果任務隊列不空,則取出任務執行,若任務隊列空,則等待?
*/??
@Override??
public?void?run()?{??
Runnable?r?=?null;??
while?(isRunning)?{//?注意,若線程無效則自然結束run方法,該線程就沒用了??
synchronized?(taskQueue)?{??
while?(isRunning??taskQueue.isEmpty())?{//?隊列為空??
try?{??
taskQueue.wait(20);??
}?catch?(InterruptedException?e)?{??
e.printStackTrace();??
}??
}??
if?(!taskQueue.isEmpty())??
r?=?taskQueue.remove(0);//?取出任務??
}??
if?(r?!=?null)?{??
r.run();//?執行任務??
}??
finished_task++;??
r?=?null;??
}??
}??
//?停止工作,讓該線程自然執行完run方法,自然結束??
public?void?stopWorker()?{??
isRunning?=?false;??
}??
}??
}
線程池通俗的描述就是預先創建若干空閑線程 等到需要用多線程去處理事務的時候去喚醒某些空閑線程執行處理任務 這樣就省去了頻繁創建線程的時間 因為頻 繁創建線程是要耗費大量的CPU資源的 如果一個應用程序需要頻繁地處理大量并發事務 不斷的創建銷毀線程往往會大大地降低系統的效率 這時候線程池就派 上用場了
本文旨在使用Java語言編寫一個通用的線程池 當需要使用線程池處理事務時 只需按照指定規范封裝好事務處理對象 然后用已有的線程池對象去自動選擇空 閑線程自動調用事務處理對象即可 并實現線程池的動態修改(修改當前線程數 最大線程數等) 下面是實現代碼
//ThreadTask java
package polarman threadpool;
/** *//**
*線程任務
* @author ryang
*
*/
public interface ThreadTask {
public void run();
}
//PooledThread java
package polarman threadpool;
import java util Collection; import 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= ; itasks length; i++) 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 Collection; import java util Iterator; import 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 = maxPoolSize; this initPoolSize = initPoolSize;
}
public void init() {
initialized = true;
for(int i= ; iinitPoolSize; i++) {
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(); isize imaxPoolSize; i++) {
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 BufferedReader; import java io IOException; import java io InputStreamReader;
import polarman threadpool ThreadPool; import 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 = taskName; this 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
標題名稱:線程池管理java代碼 java線程池的工作原理
鏈接URL:http://vcdvsql.cn/article26/doiegjg.html
成都網站建設公司_創新互聯,為您提供外貿網站建設、電子商務、建站公司、品牌網站制作、商城網站、網站設計公司
聲明:本網站發布的內容(圖片、視頻和文字)以用戶投稿、用戶轉載內容為主,如果涉及侵權請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網站立場,如需處理請聯系客服。電話:028-86922220;郵箱:631063699@qq.com。內容未經允許不得轉載,或轉載時需注明來源: 創新互聯