Java 并发编程之任务取消（七）

只运行一次的服务

以检查是否有新邮件为栗

采用AtomicBoolean而不使用Volatile类型的boolean.是因为能从内部的Runnable中访问hasNewMail标志，因此它必须是final类型以免被修改。

boolean checkMain(Set<String> hosts, long timeout, TimeUnit unit) throws InterruptedException {
		ExecutorService exec = Executors.newCachedThreadPool();
		final AtomicBoolean hasNewMail = new AtomicBoolean(false);
		try {
			for (String host : hosts) {
				exec.execute(new Runnable() {

					@Override
					public void run() {
						// TODO Auto-generated method stub
						if (checkMail(host)) {
							hasNewMail.set(true);
						}
					}
				});
			}
		} finally {
			exec.shutdown();
			exec.awaitTermination(timeout, unit);
		}
		return hasNewMail.get();
	}
           

ShutdownNow的局限性

上次说shutdownNow会返回所有尚未启动的Runnable。但是它无法返回的是正在执行的Runnable。通过封装ExecutorService并使得execute记录哪些任务是在关闭后取消的。并且在任务返回时必须保持线程的中断状态

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

public class TrackingExecutor extends AbstractExecutorService {
	private final ExecutorService exec;
	private final Set<Runnable> tasksCancelledAtShutdown = Collections
			.synchronizedSet(new HashSet<Runnable>());

	public TrackingExecutor(ExecutorService exec) {
		super();
		this.exec = exec;
	}

	public List<Runnable> getCancelledTasks() {
		if (!exec.isTerminated()) {
			throw new IllegalStateException("terminated");
		}
		return new ArrayList<Runnable>(tasksCancelledAtShutdown);
	}

	@Override
	public void execute(final Runnable command) {
		// TODO Auto-generated method stub
		exec.execute(new Runnable() {
			public void run() {
				try {
					command.run();
				} finally {
					if (isShutdown() && Thread.currentThread().isInterrupted()) {
						tasksCancelledAtShutdown.add(command);
					}
				}
			}
		});

	}

	@Override
	public void shutdown() {
		// TODO Auto-generated method stub

	}

	@Override
	public List<Runnable> shutdownNow() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public boolean isShutdown() {
		// TODO Auto-generated method stub
		return false;
	}

	@Override
	public boolean isTerminated() {
		// TODO Auto-generated method stub
		return false;
	}

	@Override
	public boolean awaitTermination(long timeout, TimeUnit unit)
			throws InterruptedException {
		// TODO Auto-generated method stub
		return false;
	}

}
           

将ExecutorService的其他方法委托给exec。以网页爬虫为例。使用TrackingExecutor的用法

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public abstract class WebCrawler {
	private volatile TrackingExecutor exec;
	private final Set<URL> urltoCrawl = new HashSet<URL>();

	public static void main(String[] args) {
		WebCrawler webc = new WebCrawler() {

			@Override
			protected List<URL> processPage(URL url) {
				System.out.println(url.getHost());
				List<URL> url2 = new ArrayList<URL>();
				try {
					url2.add(new URL("http://www.baidu.com"));
				} catch (MalformedURLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				return url2;
			}
		};
		try {
			webc.urltoCrawl.add(new URL("http://www.baidu.com"));
		} catch (MalformedURLException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		webc.start();

		try {
			Thread.sleep(1000);
			webc.stop();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public synchronized void start() {
		exec = new TrackingExecutor(Executors.newCachedThreadPool());
		for (URL url : urltoCrawl) {
			submitCrawlTask(url);
		}
		urltoCrawl.clear();
	}

	public synchronized void stop() throws InterruptedException {
		try {
			saveUncrawled(exec.shutdownNow());
			if (exec.awaitTermination(2000, TimeUnit.MILLISECONDS)) {
				saveUncrawled(exec.getCancelledTasks());
			}
		} finally {
			exec = null;
		}
	}

	protected abstract List<URL> processPage(URL url);

	private void saveUncrawled(List<Runnable> uncrawled) {
		for (Runnable runnable : uncrawled) {
			urltoCrawl.add(((CrawlTask) runnable).getPage());
		}
	}

	private void submitCrawlTask(URL u) {
		exec.execute(new CrawlTask(u));
	}

	private class CrawlTask implements Runnable {
		private final URL url;

		public CrawlTask(URL url) {
			super();
			this.url = url;
		}

		@Override
		public void run() {
			// TODO Auto-generated method stub
			for (URL link : processPage(url)) {
				if (Thread.currentThread().isInterrupted())
					return;
				submitCrawlTask(link);
			}
		}

		public URL getPage() {
			return url;
		}
	}
}
           

上面程序中抽象方法processPage中的代码是测试用的。里面正确的代码目的是获取该url下所有的链接。其实也不难。但是不属于本篇知识点内。只要得到所有<a href就可以了。有兴趣的可以看看元素选择器 还有一个开源的爬虫Crawl研究一下。

执行Stop方法后会调用saveUncrawle方法获取正在执行中的任务。之后要进行保存到文件中还是怎么办只要把代码添加到saveUncrawled内即可。