Java并发编程实践 目录
并发编程 04—— 闭锁CountDownLatch 与 栅栏CyclicBarrier
并发编程 06—— CompletionService : Executor 和 BlockingQueue
并发编程 10—— 任务取消 之 关闭 ExecutorService
并发编程 12—— 任务取消与关闭 之 shutdownNow 的局限性
并发编程 13—— 线程池的使用 之 配置ThreadPoolExecutor 和 饱和策略
概述
第1 部分 问题引入
当通过 shutdownNow 来强行关闭 ExecutorService 时，它会尝试取消正在执行的任务，并返回所有已提交但尚未开始的任务，从而将这些任务写入日志或者保存起来以便之后进行处理。
然而，我们无法通过常规方法来找出哪些任务已经开始但尚未结束。这意味着这我们无法在关闭过程中知道正在执行的任务的状态，除非任务本身会执行某种检查。要知道哪些任务还没有完成，你不仅需要知道哪些任务还没有开始，而且还需知道当 Executor 关闭时哪些任务正在执行。
第2 部分 实例
在下面程序 TrackingExecutor 中给出了如何在关闭过程中判断正在执行的任务。通过封装 ExecutorService 并使得execute 记录哪些任务是在关闭后取消的，TrackingExecutor 可以找出哪些任务已经开始但还没有正常完成。在 Executor 结束后，getCancelledTasks 返回被取消的任务清单。
/**
* 7.21 在 ExecutorService 中跟踪在关闭之后取消的任务
* @ClassName: TrackingExecutor
* @author xingle
* @date 2014-11-12 下午8:39:33
*/
public class TrackingExecutor extends AbstractExecutorService{
private final ExecutorService exec;
private final Set<Runnable> tasksCancelledAtShutdown = Collections
.synchronizedSet(new HashSet<Runnable>()); public TrackingExecutor(ExecutorService exec){
this.exec = exec;
} public List<Runnable> getCancelledTasks(){
if(!exec.isTerminated())
throw new IllegalStateException();
return new ArrayList<Runnable>(tasksCancelledAtShutdown);
} /**
*
* @Description: TODO
* @param command
* @author xingle
* @data 2014-11-13 上午9:06:56
*/
@Override
public void execute(final Runnable runnable) {
exec.execute(new Runnable() { @Override
public void run() {
try{
runnable.run();
}finally{
if(isShutdown() && Thread.currentThread().isInterrupted())
tasksCancelledAtShutdown.add(runnable);
}
}
});
} /**
* 下面将ExecutorService 的其他方法委托给 exec
*/ /**
*
* @Description: TODO
* @author xingle
* @data 2014-11-13 上午9:06:56
*/
@Override
public void shutdown() {
exec.shutdown();
} /**
*
* @Description: TODO
* @return
* @author xingle
* @data 2014-11-13 上午9:06:56
*/
@Override
public List<Runnable> shutdownNow() {
return exec.shutdownNow();
} /**
*
* @Description: TODO
* @return
* @author xingle
* @data 2014-11-13 上午9:06:56
*/
@Override
public boolean isShutdown() {
return exec.isShutdown();
} /**
*
* @Description: TODO
* @return
* @author xingle
* @data 2014-11-13 上午9:06:56
*/
@Override
public boolean isTerminated() {
return exec.isTerminated();
} /**
*
* @Description: TODO
* @param timeout
* @param unit
* @return
* @throws InterruptedException
* @author xingle
* @data 2014-11-13 上午9:06:56
*/
@Override
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
return exec.awaitTermination(timeout, unit);
} }
在程序 WebCrawler 中给出了 TrackingExecutor 的用法。网页爬虫程序的工作通常是无穷尽的，因此当爬虫程序必须关闭时，我们通常希望保持它的状态，以便稍后重启动。CrawlTask 提供了一个 getPage 方法，该方法能找出正在处理的页面。当爬虫程序关闭时，无论是还没有开始的任务，还是那些被取消的任务，都将记录他们的URL，因此当爬虫程序程序启动时，就可以将这些URL 的页面抓取任务加入到任务队列中。
/**
* 7.22 使用TrackingExecutorService 来保存未完成的任务以备后续执行
* @ClassName: WebCrawler
* TODO
* @author xingle
* @date 2014-11-13 上午9:17:54
*/
public abstract class WebCrawler {
private volatile TrackingExecutor exec;
@GuardedBy("this")
public final Set<URL> urlsToCrawl = new HashSet<URL>(); private final ConcurrentMap<URL, Boolean> seen = new ConcurrentHashMap<URL, Boolean>();
private static final long TIMEOUT = 500;
private static final TimeUnit UNIT = TimeUnit.MICROSECONDS; public WebCrawler(URL startUrl){
urlsToCrawl.add(startUrl);
} public synchronized void start(){
exec = new TrackingExecutor(Executors.newCachedThreadPool());
for (URL url: urlsToCrawl)
submitCrawlTask(url);
urlsToCrawl.clear();
} /**
* 提交爬虫任务
* @param url
* @author xingle
* @data 2014-11-13 上午9:46:01
*/
private void submitCrawlTask(URL url) {
exec.execute(new CrawlTask(url));
} protected abstract List<URL> processPage(URL url); /**
* 保存未完成的
* @param urlsToCrawl
* @author xingle
* @data 2014-11-13 上午10:10:07
*/
private void saveUncrawled(List<Runnable> uncrawled) {
for (Runnable task:uncrawled){
URL url = ((CrawlTask)task).getPage();
System.out.println("保存未完成的URL："+url);
urlsToCrawl.add(url);
} } //爬虫任务
private class CrawlTask implements Runnable{
private final URL url; CrawlTask(URL url){
this.url = url;
} private int count = 1; boolean alreadyCrawled() {
return seen.putIfAbsent(url, true) != null;
} void markUncrawled() {
seen.remove(url);
System.out.printf("marking %s uncrawled%n", url);
} @Override
public void run() {
for (URL link :processPage(url)){
if(Thread.currentThread().isInterrupted())
return;
System.out.println("提交的爬虫url:"+link);
submitCrawlTask(link);
}
} public URL getPage(){
return url;
}
} public synchronized void stop() throws InterruptedException{
try {
saveUncrawled(exec.shutdownNow());
if (exec.awaitTermination(100, UNIT)){
saveUncrawled(exec.getCancelledTasks());
} } finally {
exec = null;
}
}
}
测试程序：
public class WebCrawler_Main {
public static void main(String[] args) throws MalformedURLException{
WebCrawler webc = new WebCrawler(new URL("http://site.baidu.com/")) {
@Override
protected List<URL> processPage(URL url) {
//获取该url下所有的链接
//这里省略了该功能
List<URL> url2 = new ArrayList<URL>();
try {
url2.add(new URL("http://www.cnblogs.com/xingele0917/"));
//url2.add(new URL("http://www.zhihu.com/"));
} catch (MalformedURLException e) {
e.printStackTrace();
}
return url2;
}
};
webc.start();
try {
Thread.sleep(10);
webc.stop();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
执行结果：