线程池(中)
Posted 带剑书生
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了线程池(中)相关的知识，希望对你有一定的参考价值。
线程池（下）
线程池超负载了怎么办？都有哪些拒绝策略？
在ThreadPoolExecutor的构造方法里有一个这样的参数， RejectedExecutionHandler 通过查看Jdk我们可以知道这是一个接口，而且jdk内置实现了四种拒绝策略，它们都是ThreadPoolExecutor的public static class。
CallerRunsPolicy策略：只要线程池未关闭，该策略直接在调用者线程中，运行当前被丢弃的任务，虽然这样并不会真的丢弃任务，但是相应的被提交任务的线程性能肯定会急剧下降。AbortPolicy策略：该策略会直接抛出异常，阻止系统正常工作。DiscardPolicy策略：直接丢弃任务，不给予任何处理。DiscardOldestPolicy策略：该策略会丢弃最古老的请求(也就是任务队列中最先进入的任务)，即将被执行的策略，并尝试再次提交当前任务。
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ExecutorServiceDemo {
static void log(String msg) {
System.out.println(System.currentTimeMillis() + " -> " + msg);
}
public static void main(String[] args) throws Exception {
ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(1));
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
for (int i = 0; i < 10; i++) {
final int index = i;
pool.submit(new Runnable() {
public void run() {
log("run task:" + index + " -> " + Thread.currentThread().getName());
try {
Thread.sleep(1000L);
} catch (Exception e) {
e.printStackTrace();
}
log("run over:" + index + " -> " + Thread.currentThread().getName());
}
});
}
log("before sleep");
Thread.sleep(4000L);
log("before shutdown()");
pool.shutdown();
log("after shutdown(),pool.isTerminated=" + pool.isTerminated());
pool.awaitTermination(1, TimeUnit.SECONDS);
log("now,pool.isTerminated=" + pool.isTerminated());
}
}
程序设置的线程池只有一个线程，并且允许最大的线程数量也为1，同时任务队列的最大界限为1。 当采用DisCardOldestPolicy()时：
可以知道在workQueue中的task1-task8一直被丢弃（因为任务队列只有一个容量，而线程池里的唯一线程处理的速度不是很快，而程序又不停的往线程池里提交任务）。直到最后一个task9任务才被执行。 当采用DisCardPolicy()时：
可以看见刚除开始线程执行的task0和任务队列里的task1，其余的都被默默地抛弃了。 当采用AbortPolicy()策略时：系统直接抛出异常。。。
当采用CallerRunsPolicy()策略时就有点不同的了：
可以看见在main线程一直帮忙处理不能被线程池处理同时也不能进入任务队列的任务。 若以上的策略还是无法满足实际应用的需要。我们还可以自己扩展
RejectedExecutionHandler {
void rejectedExecution(Runnable r,ThreadPoolExecutor executor);
}
自己定义线程创建：ThreadFactory
看到这里我们可能有疑问那就是线程池的线程是从哪里来的？答案就是：ThreadFactory 它是一个接口，只有一个方法
public Thread newThread(Runnable r);
自定义线程池可以更加自由的设置线程的状态
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* 自定义线程池 :public interface ThreadFactory
* 根据需要创建新线程的对象
* Thread newThread(Runnable r)
* @author Administrator
*
*/
public class MyThreadFactory {
public static class MyTask implements Runnable {
@Override
public void run() {
System.out.println( System.currentTimeMillis() + " Thread id:" + Thread.currentThread().getId());
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) throws InterruptedException {
MyTask task = new MyTask();
ExecutorService exec = new ThreadPoolExecutor(5,5,0L,TimeUnit.MILLISECONDS,
new SynchronousQueue<Runnable>(),
new ThreadFactory(){
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r,"后台线程");
t.setDaemon(true);
System.out.println("create " + t.getName()+ " " + t.getId());
return t;
}
});
for(int i = 0; i < 5; i++) {
exec.submit(task);
}
TimeUnit.SECONDS.sleep(2);
}
}
实际中使用guava的ThreadFactoryBuilder来创建一个ThreadFactory更加灵活方便。
线程池的扩展
上面仅仅只是我们自定义了创建线程时的状态，但是有时候，我们需要对线程执行的任务进行监控，比如说任务的开始时间和结束时间。幸运的是，ThreadPoolExecutor是一个可扩展的线程池。提供了三个方法：
protected void beforeExecute(Thread t, Runnable r) { }
protected void afterExecute(Runnable r, Throwable t) { }
protected void terminated() { }
同时再看ThreadPoolExecutor中有一个这样的类， private final class Worker extends AbstractQueuedSynchronizer implements Runnable它里面有一个方法runWorker
final void runWorker(Worker w) {
Runnable task = w.firstTask;
w.firstTask = null;
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
clearInterruptsForTaskRun();
try {
beforeExecute(w.thread, task);//运行前
Throwable thrown = null;
try {
task.run();//运行任务
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);//运行结束后
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
ThreadPoolExecutor中工作线程正是Worker的实例（它是把Runnable对象进行了包装），Worker.runWorker()会被线程池以多线程模式异步调用，即它会被多个线程访问，因此其beforeExecute()它们也会被多线程同时访问。 在默认的ThreadExecutor实现中，提供了beforeExecute()，afterExecute()，terminated()空的实现。在实际的应用中可以通过对其进行扩展实现多线程池运行状态的跟踪。
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ExtThreadPool {
public static class MyTask implements Runnable {
private String name;
public MyTask(String name) {
this.name = name;
}
@Override
public void run() {
System.out.println("正在执行线程id：" + Thread.currentThread().getId() + " Task name " + name);
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) throws InterruptedException {
ExecutorService exec =
new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>())
//直接获得ThreadPoolExecutor的子类，并且重写protect钩子方法
{
@Override
protected synchronized void beforeExecute(Thread t, Runnable r) {
System.out.print(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SS").format(new Date()));
System.out.println("准备执行：" + ((MyTask)r).name);
}
@Override
protected synchronized void afterExecute(Runnable r, Throwable t) {
System.out.print(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SS").format(new Date()));
System.out.println("执行完成：" + ((MyTask)r).name);
}
@Override
protected synchronized void terminated() {
System.out.print(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SS").format(new Date()));
System.out.println("线程池退出...");
}
};
for(int i = 0; i < 5; i++) {
MyTask task = new MyTask("Task" + i);
exec.execute(task);
Thread.sleep(10);
}
exec.shutdown();
}
}
这里需要注意打印时间这句会发生线程安全。用Synchronized监视器让线程在临界区进行互斥的执行。 运行结果：
这里可以清楚地看到线程池中执行任务的创建和结束信息。
关于我
以上是关于线程池(中)的主要内容，如果未能解决你的问题，请参考以下文章