Java Thread Pool 例子
2012-11-16 15:33
218 查看
http://muscle-liu.iteye.com/blog/328252
http://blog.csdn.net/v1v1wang/article/details/5607279
虽然从 Java5 开始 JDK 里的 java.util.concurrent 包内建了线程池,你不必自己实现线程池,但理解线程的实现原理对 Java 编程很有用。
当你想把运行在你的程序中的线程控制在一定的数量之内,线程池就显得非常有用。
引用
原理:
用一个阻塞队列里(Blocking Queue)来存储线程池的所有空闲线程。不用为每个任务都创建一个新的线程,可以把任务当参数传到线程池里。只要当线程池有空闲的线程时,这个任务就会被执行。
BlockingQueue :
ThreadPool :
-- BlockingQueue
-- threads <TaskThread>
TaskThread :
-- BlockingQueue (shared by all taskThread)
BlockQueue.java
Java代码
public class BlockingQueue {
private List queue = new LinkedList();
private int limit = 10;
public BlockingQueue(int limit) {
this.limit = limit;
}
public synchronized Object enqueue(Object item)
throws InterruptedException {
while (this.queue.size() == this.limit) {
System.out.println("Arrive to the pool's max size:"+this.limit);
wait();
}
if (this.queue.size() == 0) {
notifyAll();
}
System.out.println("Add a new thread to the pool.Pool's size before
adding:"+this.queue.size());
this.queue.add(item);
return this.queue.get(this.queue.size()-1);
}
public synchronized Object dequeue()
throws InterruptedException {
while (this.queue.size() == 0) {
System.out.println("Pool's size is 0.");
wait();
}
if (this.queue.size() == this.limit) {
notifyAll();
}
System.out.println("Remove a thread from the pool.Pool's size before
removing:"+(this.queue.size()-1));
return this.queue.remove(0);
}
public synchronized int size(){
return this.queue.size();
}
}
当阻塞队列没达到界限值(最大值与最小值)时,插入与出列正常,没限制。阻塞队列达到最大值时,再想插入一个线程,队列就会停止操作,让插入在等待,同时唤醒出列操作。相反,当全部出列后,再想出列时,出列操作就会停止,在等待,同时唤醒插入操作。
线程池通常用在多线程服务端的,下边是实现线程池的其他类,其中:
Java代码
/**
* Implements the thread pool
* @author winxp
*/
public class ThreadPool {
private BlockingQueue taskQueue = null;
private List<TaskThread> threads = new ArrayList<TaskThread>();
private boolean isStopped = false;
/**
* Create a new ThreadPool
* @param noOfThreads Initiate num of threads in the thread pool线程池的初始化线程数
* @param maxNoOfTasks Max num of threads in the thread pool线程池的最大线程数
*/
public ThreadPool(int noOfThreads, int maxNoOfTasks) {
taskQueue = new BlockingQueue(maxNoOfTasks);
for (int i = 0; i < noOfThreads; i++) {
threads.add(new TaskThread(taskQueue));
}
for (TaskThread thread : threads) {
thread.start();
}
}
/**
*
* @param task
*/
public synchronized void execute(Runnable task){
if(this.isStopped)throw
new IllegalStateException("ThreadPool is stopped");
try {
this.taskQueue.enqueue(task);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
/**
*
*/
public synchronized void stop() {
this.isStopped = true;
for (TaskThread thread : threads) {
thread.interrupt();
}
}
}
/**
* Simulate that dequeue from the BlockingQueue, and execute the dequeuing task
*/
public class TaskThread extends Thread {
private BlockingQueue taskQueue = null;
private boolean isStopped = false;
public TaskThread(BlockingQueue queue) {
taskQueue = queue;
}
@Override
public void run() {
while (!isStopped()) {
try {
Runnable runnable = (Runnable) taskQueue.dequeue();
runnable.run();
} catch (Exception e) {
//log or otherwise report exception,
//but keep pool thread alive.
}
}
}
@Override
public synchronized void interrupt() {
isStopped = true;
super.interrupt();//break pool thread out of dequeue() call.
}
public synchronized boolean isStopped() {
return isStopped;
}
}
测试类:
Java代码
/**
* Testing class
*/
class TestThread extends Thread{
private int num;
public TestThread(int num){
this.num = num;
}
@Override
public void run(){
System.out.println("TestThread index: "+num);
try {
Thread.sleep(1000);
} catch (InterruptedException ex) {
}
}
/**
*
* @param args
*/
public static void main(String[] args){
ThreadPool tp = new ThreadPool(3,10);
for(int i = 0;i<15;i++){
TestThread test = new TestThread(i);
tp.execute(test);
}
}
}
http://blog.csdn.net/v1v1wang/article/details/5607279
虽然从 Java5 开始 JDK 里的 java.util.concurrent 包内建了线程池,你不必自己实现线程池,但理解线程的实现原理对 Java 编程很有用。
当你想把运行在你的程序中的线程控制在一定的数量之内,线程池就显得非常有用。
引用
原理:
用一个阻塞队列里(Blocking Queue)来存储线程池的所有空闲线程。不用为每个任务都创建一个新的线程,可以把任务当参数传到线程池里。只要当线程池有空闲的线程时,这个任务就会被执行。
BlockingQueue :
ThreadPool :
-- BlockingQueue
-- threads <TaskThread>
TaskThread :
-- BlockingQueue (shared by all taskThread)
BlockQueue.java
Java代码
public class BlockingQueue {
private List queue = new LinkedList();
private int limit = 10;
public BlockingQueue(int limit) {
this.limit = limit;
}
public synchronized Object enqueue(Object item)
throws InterruptedException {
while (this.queue.size() == this.limit) {
System.out.println("Arrive to the pool's max size:"+this.limit);
wait();
}
if (this.queue.size() == 0) {
notifyAll();
}
System.out.println("Add a new thread to the pool.Pool's size before
adding:"+this.queue.size());
this.queue.add(item);
return this.queue.get(this.queue.size()-1);
}
public synchronized Object dequeue()
throws InterruptedException {
while (this.queue.size() == 0) {
System.out.println("Pool's size is 0.");
wait();
}
if (this.queue.size() == this.limit) {
notifyAll();
}
System.out.println("Remove a thread from the pool.Pool's size before
removing:"+(this.queue.size()-1));
return this.queue.remove(0);
}
public synchronized int size(){
return this.queue.size();
}
}
当阻塞队列没达到界限值(最大值与最小值)时,插入与出列正常,没限制。阻塞队列达到最大值时,再想插入一个线程,队列就会停止操作,让插入在等待,同时唤醒出列操作。相反,当全部出列后,再想出列时,出列操作就会停止,在等待,同时唤醒插入操作。
线程池通常用在多线程服务端的,下边是实现线程池的其他类,其中:
Java代码
/**
* Implements the thread pool
* @author winxp
*/
public class ThreadPool {
private BlockingQueue taskQueue = null;
private List<TaskThread> threads = new ArrayList<TaskThread>();
private boolean isStopped = false;
/**
* Create a new ThreadPool
* @param noOfThreads Initiate num of threads in the thread pool线程池的初始化线程数
* @param maxNoOfTasks Max num of threads in the thread pool线程池的最大线程数
*/
public ThreadPool(int noOfThreads, int maxNoOfTasks) {
taskQueue = new BlockingQueue(maxNoOfTasks);
for (int i = 0; i < noOfThreads; i++) {
threads.add(new TaskThread(taskQueue));
}
for (TaskThread thread : threads) {
thread.start();
}
}
/**
*
* @param task
*/
public synchronized void execute(Runnable task){
if(this.isStopped)throw
new IllegalStateException("ThreadPool is stopped");
try {
this.taskQueue.enqueue(task);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
/**
*
*/
public synchronized void stop() {
this.isStopped = true;
for (TaskThread thread : threads) {
thread.interrupt();
}
}
}
/**
* Simulate that dequeue from the BlockingQueue, and execute the dequeuing task
*/
public class TaskThread extends Thread {
private BlockingQueue taskQueue = null;
private boolean isStopped = false;
public TaskThread(BlockingQueue queue) {
taskQueue = queue;
}
@Override
public void run() {
while (!isStopped()) {
try {
Runnable runnable = (Runnable) taskQueue.dequeue();
runnable.run();
} catch (Exception e) {
//log or otherwise report exception,
//but keep pool thread alive.
}
}
}
@Override
public synchronized void interrupt() {
isStopped = true;
super.interrupt();//break pool thread out of dequeue() call.
}
public synchronized boolean isStopped() {
return isStopped;
}
}
测试类:
Java代码
/**
* Testing class
*/
class TestThread extends Thread{
private int num;
public TestThread(int num){
this.num = num;
}
@Override
public void run(){
System.out.println("TestThread index: "+num);
try {
Thread.sleep(1000);
} catch (InterruptedException ex) {
}
}
/**
*
* @param args
*/
public static void main(String[] args){
ThreadPool tp = new ThreadPool(3,10);
for(int i = 0;i<15;i++){
TestThread test = new TestThread(i);
tp.execute(test);
}
}
}
相关文章推荐
- JAVA:谈谈Thread Pool
- 8、java5线程池之动态缓存线程池newCachedThreadPool
- Java threadpool机制深入分析
- Java threadpool机制深入分析
- Caused by: java.lang.NoClassDefFoundError: Could not initialize class org.elasticsearch.threadpool.ThreadPool
- Java threadpool机制深入分析
- Java四种线程池newCachedThreadPool,newFixedThreadPool,newScheduledThreadPool,newSingleThreadExecutor
- java 线程池 ---- Eclipse自带的threadPool
- Socket + ThreadPool的例子
- Java-线程池 ThreadPool 专题详解 (美团面试题)
- java ThreadPool 线程池
- java 通过Executors.newFixedThreadPool创建固定大小的线程执行器
- Java Thread Pool Example using Executors and ThreadPoolExecutor
- Java - 什么是线程池(thread pool)?
- java threadPool 线程池简单分析
- Java Thread Pool
- Java threadpool机制深入分析
- Java threadpool机制深入分析
- MinerThreadPool.java 线程池
- Java newFixedThreadPool监控当前线程的运行状况