Java多线程之ThreadPoolExecutor

1.ExecutorService demo

package com.ccy.thread.demo;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
*
* <p>
* Title: ThreadPoolTest.java
* Package com.ccy.thread.demo
* </p>
* <p>
* Description: 线程池简单demo
* <p>
* @author Tom.Cai
* @created 2015-12-30 下午10:45:14
* @version V1.0
*
*/
public class ThreadPoolTest {
public static void main(String[] args) {
//创建一个可重用固定线程数的线程池，以共享的无界队列方式来运行这些线程。
ExecutorService threadPool = Executors.newFixedThreadPool(3);
for (int i = 0; i < 6; i++) {
Runnable worker = new workThread(String.valueOf(i));
threadPool.execute(worker);
}
//启动一次顺序关闭，执行以前提交的任务，但不接受新任务。
threadPool.shutdown();
//  如果关闭后所有任务都已完成，则返回 true。
while (!threadPool.isTerminated()) {

}
System.out.println("all threads Finished");
}
}

class workThread implements Runnable{
private String command;

public workThread(String command){
this.command = command;
}

@Override
public void run() {
System.out.println(Thread.currentThread().getName()+" Start time-"+command);
try {
Thread.sleep(5000);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName()+" End !!!");
}

}

2.ThreadPoolExecutor demo

package com.ccy.thread.demo;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
*
* <p>
* Title: ThreadPoolExecutorTest.java
* Package com.ccy.thread.demo
* </p>
* <p>
* Description: 线程池监控demo
* <p>
* @author Tom.Cai
* @created 2015-12-30 下午10:48:21
* @version V1.0
*
*/
public class ThreadPoolExecutorTest {
public static void main(String[] args) throws Exception {
RejectedExecutionHandlerImpl rejectionHandler = new RejectedExecutionHandlerImpl();
ThreadFactory threadFactory = Executors.defaultThreadFactory();
//在初始化 ThreadPoolExecutor 时，初始线程池大小设为2、最大值设为4、线程工作等待队列大小设为2。
//所以，如果当前有4个任务正在运行而此时又有新任务提交，工作队列将只存储2个任务和其他任务将交由RejectedExecutionHandlerImpl 处理。
ThreadPoolExecutor executorPool = new ThreadPoolExecutor(2, 4, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2), threadFactory, rejectionHandler);
MonitorThread monitor = new MonitorThread(executorPool);
Thread monitorThread = new Thread(monitor);
monitorThread.start();
for(int i=0; i<10; i++){
executorPool.execute(new workerThread(String.valueOf(i)));
}
Thread.sleep(30000);
executorPool.shutdown();
Thread.sleep(5000);
monitor.shutdown();
}
}

class RejectedExecutionHandlerImpl implements RejectedExecutionHandler {

@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.out.println(r.toString() + " is rejected");
}

}

class MonitorThread implements Runnable{
private ThreadPoolExecutor executor;
private boolean runFlag = true;

public MonitorThread(ThreadPoolExecutor executor){
this.executor = executor;
}

public void shutdown(){
this.runFlag = false;
}

@Override
public void run() {
while(runFlag){
System.out.println(
String.format("[monitor] [%d/%d] Active: %d, Completed: %d, Task: %d, isShutdown: %s, isTerminated: %s",
// 返回池中的当前线程数。
this.executor.getPoolSize(),
// 返回核心线程数。
this.executor.getCorePoolSize(),
this.executor.getActiveCount(),
this.executor.getCompletedTaskCount(),
//返回曾计划执行的近似任务总数。
this.executor.getTaskCount(),
this.executor.isShutdown(),
this.executor.isTerminated()));
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

}

class workerThread implements Runnable{
private String command;

public workerThread(String command){
this.command = command;
}

@Override
public void run() {
System.out.println(Thread.currentThread().getName()+" Start time-"+command);
try {
Thread.sleep(5000);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName()+" End !!!");
}

}