文档章节

java并发编程(六)取消与关闭

呆萌的我
 呆萌的我
发布于 2015/10/13 20:11
字数 2884
阅读 8
收藏 0

接《java并发编程(五)任务执行


前面几章我们一直是创建和开启线程,而有时候我们要结束任务或线程,这并不是很容易的,因为,java 并没有提供任何机制来安全终止线程(在未来的jdk版本中会不会加入呢?)它提供了中断。这是一种机制,能够在一个线程终止另一个线程的工作。

任务取消

一种协作方式是设置某个“已请求取消”标识,而任务将定时查看该标志。下面程序,将持续枚举素数,直到它将被取消。cancle方法设置canceled标志,并且主循环在搜索下一个素数下一个之前,首先检查这个标志(canceled必须为volatile 原因请看 java内存模型
public class PrimeGenerator implements Runnable {

	private List<BigInteger> primes = new ArrayList<BigInteger>();
	private volatile boolean canceled;

	@Override
	public void run() {
		BigInteger bigInteger = BigInteger.ONE;
		while (!canceled) {
			bigInteger = bigInteger.nextProbablePrime();
			synchronized (this) {
				primes.add(bigInteger);
			}
		}
	}

	public void cancel() { this.canceled = true; }

	public synchronized List<BigInteger> get() {
		return new ArrayList<BigInteger>(primes);
	}
}
现在我们要让一个素数生成器运行1秒后停止,(虽然不是很精确,因为每一条代码是有延时的),我们可以用try finally 来完成,以保证线程会停止,否则线程会一直消耗CPU时钟周期,导致JVM退出。
List<BigInteger> aSecondOfPrimes() throws InterruptedException {
	
		PrimeGenerator generator = new PrimeGenerator();
		new Thread(generator).start();
		try {
			Thread.sleep(1000);
		} finally {
			generator.cancel();
		}
		return generator.get();
	}

 中断

PrimeGenerator中的取消机制最终会使得搜索的素数任务退出,但是退出过程中需要花费一定的时间,然而,如果任务调度了一些阻塞方法,(BlockingQueue.put)那么可能产生一个问题——任务可能永远不会检查取消标示,永远不会结束。
为取保线程能退出,我们通常使用中断,当我们调用interrupt,并不意味着立即停止目标线程正在运行的线程,而只是传递了一个请求中断的信息,它会在线程下一个合适的的时刻中断自己。wait、sleep、join、将严格处理这种请求,当他们收到一个中断请求,或饿着开始执行时发现中断状态时,将抛出异常。
使用静态的interrupted时应该小心,因为它会清除当前线程的中断状态,如果返回true,除非你你想屏蔽这个中断,否则必须对它进行处理,抛出异常或者再次调用interrupt来恢复中断状态
try {
			...
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}
通常,我们用中断来取消任务比检查标记更好,是最合适的取消任务方式,我们看一个更加健壮的获得素数的类。
public class PrimeProducer extends Thread{
	
	private final BlockingQueue<BigInteger> queue;
	
	public PrimeProducer(BlockingQueue<BigInteger> queue) {
		this.queue = queue;
	}

	@Override
	public void run() {
		try {
			BigInteger p = BigInteger.ONE;
			while(!Thread.currentThread().isInterrupted()) //①用线程的状态来检查
				queue.put(p = p.nextProbablePrime());
			
		} catch (InterruptedException e) {
			//中断将线程退出
		}
	}
	public void cancel() { interrupt(); }

}

我们分析下,在while时,我们有两次的检查中断,while中有一次,在执行put的时候有一次,这样我们比用flag标识有更高的效用性,通常,我们也是通过这种方式来取消线程的。

通过Future来实现中断

public static void timeRun(Runnable r ,long timeout, TimeUnit unit) {
		
		Future<?> task = taskExec.submit(r);
		
		try {
			task.get(timeout,unit);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ExecutionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (TimeoutException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			task.cancel(true);
		}
		
	}

处理不可中断的请求

java中并非所有的可阻塞方法或者阻塞机制都能响应中断,如果一个线程由于执行同步的Socket I/O或者等待获得内置锁而阻塞,那么中断请求只能设置线程的中断状态,除此之外,没有任何作用,对于那些由于执行不可中断的操作而被阻塞的线程,可以使用类似于中断手段来停止这些线程,但是要求我们必须知道线程阻塞的原因。
  • java.io包中的同步Socket I/O。
  • java.io包中的同步I/O
  • Select的异步I/O
  • 获取某个锁
public class ReadThread extends Thread {

	private static final int BUFSZ = 1024;
	private final Socket socket;
	private final InputStream in;

	public ReadThread(Socket socket, InputStream in) {
		this.socket = socket;
		this.in = in;
	}

	@Override
	public void interrupt() {

		try {
			socket.close();
		} catch (IOException e) {
		} finally {
			super.interrupt();
		}
	}

	@Override
	public void run() {
		byte[] buf = new byte[BUFSZ];
		int count;
		try {
			while (true) {
				count = in.read(buf);
				if (count < 0)
					break;
				else if (count > 0)
					processBuffer(buf, count);
			}
		} catch (Exception e) {

		}
	}

}
我们看到 如果只终止ReadThread线程是没用的,socket是不能关闭的,所以我们要重写interrupt方法 将socket关闭。当然还要调用父类的interrupt

采用newTaskFor来封装非标准的取消

我们通过newTaskFor方法进一步优化ReadThread中封装非标准取消的技术,ThreadPoolExecutor是java 6新增功能,当把一个Callable提交给ExecutorService时,submit方法返回一个Future,我们可以通过Future来取消任务。newTaskFor是一个工厂方法,它将创建一个Future来代表任务。newTaskFor还能返回一个RunnableFuture借口,该接口扩展了Future和Runnable。
通过定制便是任务的Future可以改变Future.cancel的行为。

首先我们继承Callable接口进行扩展
interface CancellableTask<T> extends Callable<T> {
	void cancel();
	RunnableFuture<T> newTask();
}

然后定义一个抽象的 SocketUsingTask继承 CancellableTask,重写 cancel(Callellable接口)方法,实现的socket关闭,重写 newTask为 CancellingExecutor类使用。
public abstract class SocketUsingTask<T> implements CancellableTask<T> {

	private Socket socket;
	protected synchronized void setSocket(Socket s) {
		socket = s;
	}

	@Override
	public synchronized void cancel() {
		try {
			if (socket != null)
				socket.close();
		} catch (Exception e) {
		}
	}

	@Override
	public RunnableFuture<T> newTask() {
		return new FutureTask<T>(this) {
			
			@Override
			public boolean cancel(boolean mayInterruptIfRunning) {
				try {
					SocketUsingTask.this.cancel();
				} catch (Exception e) {
				} finally {
					//不建议将return 写在finally中
				}
				return super.cancel(mayInterruptIfRunning);
			}
		};
	}
}

现在我们继承 ThreadPoolExecutor,写一个工厂类( CancellingExecutor )
public class CancellingExecutor extends ThreadPoolExecutor {

	public CancellingExecutor(int corePoolSize, int maximumPoolSize,
			long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
	}

	@Override
	protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {

		if (callable instanceof CancellableTask)
			return ((CancellableTask<T>) callable).newTask();
		else
			return super.newTaskFor(callable);
	}

}

ok 现在我们测试
public class TTT {
	public static void main(String[] args) {
		
		CancellingExecutor cancellingExecutor = new CancellingExecutor(10, 20, 200, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10));
		Future<String> future = cancellingExecutor.submit(new SocketUsingTask<String>() {
			@Override
			public String call() throws Exception {
				return "hello";
			}
		});
		
		future.cancel(true);
		
		
	}
}


停止基于线程的服务

对于持有线程的服务,只要服务的存在时间大于创建线程的方法的存在时间,那么就应该提供生命周期方法。


public class LogService {

	private final BlockingQueue<String> queue;
	private final LoggerThread loggerThread;
	private final PrintWriter printWriter;
	private volatile boolean isShutdown;
	private int reservation;

	public LogService(BlockingQueue<String> queue, boolean isShutdown,
			PrintWriter printWriter) {
		this.queue = queue;
		this.loggerThread = new LoggerThread();
		this.printWriter = printWriter;
	}

	public void start() {
		loggerThread.start();
	}

	public void stop() {
		synchronized (this) {
			isShutdown = true;
		}
		loggerThread.interrupt();
	}

	public void put(String msg) throws InterruptedException {
		synchronized (this) {
			if (isShutdown)
				throw new IllegalStateException();
			++reservation;
		}
		queue.put(msg);
	}

	private class LoggerThread extends Thread {

		@Override
		public void run() {
			try {
				while (true) {
					try {
						synchronized (LogService.this) {
							if (isShutdown && reservation == 0)
								break;
						}
						String msg = queue.take();
						printWriter.write(msg);
						synchronized (LogService.this) {
							--reservation;
						}

					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}

			} finally {
				// 接到interrupt会执行
				printWriter.close();
			}
		}
	}

}

关闭ExecutorService

ExecutorService提供两个关闭的方法,shutdown 和shutdownNow。shutdownNow首先关闭当前执行线程,然后返回未执行的任务,而shutdown则等待执行完成。两种方法差别在于安全性和响应性。
直接看例子了,这没什么好说的。

public class LogService {

	private static final TimeUnit UNIT = null;
	private static final long TIMEOUT = 0;
	private final ExecutorService exec = newSingleThreadExecutor();
	
	public void start() { }
	
	public void stop() {
		try {
			exec.shutdown();
			exec.awaitTermination(TIMEOUT, UNIT);
		} finally {
			writer.close();
		}
	}
	
	public void log(String msg) {
		try {
			exec.submit(new WriteTask(msg));
		} catch (Exception e) { }
	}
	
}

毒丸对象

另一种关闭生产者-消费者服务的方式就是使用“毒丸”对象,其实就是指往对象里面放一个标志对象,当得到这个对象就立即停止,这就需要在执行方法里面判断,消费者读到毒丸后就不会再执行,同样生产者提交毒丸后,就不能再提交任务。
只有生产者和消费者都已知的情况下,才可以使用“毒丸”,当生产者和消费者和数量较大时,方法变的难以使用。

直接看书中的例子,这个不是很常用,粘过来,以备用吧。

public class IndexingService {
	private static final File POISON = new File("");
	private final IndexerThread consumer = new IndexerThread();
	private final CrawlerThread producer = new CrawlerThread();
	private final BlockingQueue<File> queue;
	private final FileFilter fileFilter;
	private final File root;

	class CrawlerThread extends Thread {
		class IndexerThread extends Thread {
			public void start() {
				producer.start();
				consumer.start();
			}
			/* Listing 7.18 */
		} /* Listing 7.19 */
	}

	public void stop() {
		producer.interrupt();
	}

	public void awaitTermination() throws InterruptedException {
		consumer.join();
	}
}

public class CrawlerThread extends Thread {
	public void run() {
		try {
			crawl(root);
		} catch (InterruptedException e) { /* fall through */
		} finally {
			while (true) {
				try {
					queue.put(POISON);
					break;
				} catch (InterruptedException e1) { /* retry */
				}
			}
		}
	}

	private void crawl(File root) throws InterruptedException { // ...
	}
}

public class IndexerThread extends Thread {
	public void run() {
        try {
            while (true) {
            	File file = queue.take();
            	if (file == POISON)
            		break;
            	else indexFile(file);
            }
        } catch(InterruptedException consumed) { }
	}
}

示例:只执行一次的服务

如果某个方法需要处理一批任务,并且当所有任务都处理完成后才返回,那么可以通过一个私有的Executor来简化服务的生命周期,其中该Executor的生命周期由这个方法来控制。
下面程序checkMail方法能在多台主机上并行地检查新邮件,它创建一个私有的Executor,并向每台主机提交一个任务,当所有邮件任务都执行完成后,关闭并结束。
boolean checkMail(Set<String> hosts, long timeout, TimeUnit unit)
			throws InterruptedException {

		ExecutorService exec = Executors.newCachedThreadPool();
		final AtomicBoolean hasNewMail = new AtomicBoolean(false);

		try {
			for (final String host : hosts) {
				exec.execute(new Runnable() {

					@Override
					public void run() {
						if (check(host))
							hasNewMail.set(true);
					}
				});
			}
		} finally {
			exec.shutdown();
			exec.awaitTermination(timeout, unit);
		}
		return hasNewMail.get();
	}

用AtomicBoolean代替volatile boolean,是因为从内部的Runnable中访问hasNewMail标志,它必须定义为final

shutdownNow的局限性

当通过shutdownNow来强行关闭ExecutorServices时,它会尝试正在执行的任务,并返回所有已提交但尚未开始的任务。而然我想得到哪些已经开始但是尚未执行完成就被打断的线程,就需要自己封装了。

 设计一个TrackingExecutor类继承AbstractExecutorService 重写execute 当执行时被中断记录在集合里。

public class TrackingExecutor extends AbstractExecutorService {

	private final ExecutorService exec;
	private final Set<Runnable> taskCancellAtShutdown = Collections.synchronizedSet(new HashSet<Runnable>());

	public TrackingExecutor(ExecutorService exec) {
		this.exec = exec;
	}
	
	
	public List<Runnable> getCancelledTask() {
		if(!exec.isTerminated())
			throw new IllegalStateException();
		return new ArrayList<Runnable>(taskCancellAtShutdown);
	}

	@Override
	public void execute(Runnable command) {
		exec.execute(new Runnable() {

			@Override
			public void run() {
				try {
					command.run();
				} finally {
					if (isShutdown() && Thread.currentThread().isInterrupted())
						taskCancellAtShutdown.add(command);
				}
			}
		});
	}

	...

}

public abstract class WebCrawler {

	private volatile TrackingExecutor exec;
	private final Set<URL> urlToCrawl = new HashSet<URL>();

	public synchronized void start() {
		exec = new TrackingExecutor(Executors.newCachedThreadPool());
		for (URL url : urlToCrawl) submitCrawlTask(url);
		urlToCrawl.clear();
	}
	
	public synchronized void stop() {
		
		try {
		saveUncrawled(exec.shutdownNow());
		if(exec.awaitTermination(timeout, unit))
			saveUncrawled(exec.getCancelledTask());
		}finally {
			exec = null;
		}
	}
	
	protected abstract List<URL> processPage(URL url);
	
	private void saveUncrawled(List<Runnable> uncrawled) {
		for (Runnable task : uncrawled) {
			urlToCrawl.add(((CrawTask)task).getUrl());
		}
	}
	
	private void submitCrawlTask(URL u) {
		exec.execute(new CrawTask(u));
	}
	
	
	private class CrawTask implements Runnable {

		private final URL url;
		
		public CrawTask(URL url) {
			this.url = url;
		}

		@Override
		public void run() {
			for (URL link : processPage(url)) {
				if(Thread.currentThread().isInterrupted())
					return;
				submitCrawlTask(link);
			}
		}

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

处理非正常的线程终止

Thread API中提供了一个UncaughtExceptionHandler,能够检测出线程由于未捕获的异常儿终结的情况。
当一个线程由于未捕获异常而终止时,JVM会把这个事件报告给应用程序提供的UncaughtExceptionHandler异常处理器。如果没有提过任何异常处理器,就会输出System.err。
class UELogger implements Thread.UncaughtExceptionHandler {

	@Override
	public void uncaughtException(Thread t, Throwable e) {
		//进行错误日志的捕获
	}
	
}
可以通过Thread. setUncaughtExceptionHandler 设置
只有通过executor提交的任务抛出的异常交给未捕获异常处理器,通过submit提交的任务,异常都会作为任务返回状态的一部分。如果一个由submit提交的任务由于跑出异常而结束,那么这个异常将被Future.get封装在ExecutorException中重新抛出。

关闭钩子

正常关闭的触发方式:
  1. 最后一个“正常(非守护)”线程结束
  2. 调用System.exit
  3. Ctrl + C

public class Hook {

	public static void main(String[] args) {
		
		Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
			
			@Override
			public void run() {
				System.out.println("!!");
			}
		}));
		
	}
}


在start方法中注册关闭钩子
public void start() {
		Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {

			@Override
			public void run() {
				//TODO
			}
		}));
	}



版权声明:本文为博主原创文章,未经博主允许不得转载。

© 著作权归作者所有

共有 人打赏支持
呆萌的我
粉丝 4
博文 15
码字总数 15443
作品 0
天津
私信 提问
java并发编程(六): 取消与关闭

取消与关闭: 如何正确,安全地取消或关闭任务。 任务取消: 若外部代码能在某个操作正常完成之前将其置入“完成”状态,则还操作是可取消的。 取消操作的原因: 1. 用户请求取消。 2. 有时间...

ihaolin
2014/03/30
0
0
读书笔记之《Java并发编程的艺术》-线程池和Executor的子孙们

读书笔记部分内容来源书出版书,版权归本书作者,如有错误,请指正。 欢迎star、fork,读书笔记系列会同步更新 git https://github.com/xuminwlt/j360-jdk module j360-jdk-thread/me.j360....

Hi徐敏
2015/11/11
0
1
Google的Guava类库简介(转)

说明:信息虽然有点旧,至少可以先了解个大概。 Guava是一个Google的基于Java的类库集合的扩展项目,包括collections, caching, primitives support, concurrency libraries, common annotat...

easonjim
2017/11/01
0
0
[转] Java 启动参数大全

前段时间系统升级时遭遇了OOM,具体解决过程见 遭遇OutOfMemoryError; 为了巩固对于java启动各项参数的认识,决定将所有参数列举出来,并一一解释,以便后查; java启动参数共分为三类; 其一...

dodojava
2011/06/08
0
1
阿里P8架构师总结的互联网Java架构系统化学习路线

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/t4i2b10X4c22nF6A/article/details/83181980 网络应用,最常见的研发语言是Java的和PHP。 后端服务,最常见的...

JAVA高级架构v
10/19
0
0

没有更多内容

加载失败,请刷新页面

加载更多

mybatis学习(1)

JDBC连接方式: 1.底层没有使用连接池,操作数据库需要频繁的创建和关闭连接,消耗资源。 2.写原生的JDBC代码在JAVA中,一旦需要修改SQL的话(比如表增加字段),JAVA需要整体重新编译,不利...

杨健-YJ
20分钟前
1
0
怎么组织文档

可以从以下几个方面考虑组织文档: ☐ 各种分支的界面截图和对应的类及文件 ☐ 框架或类图 ☐ 流程图 ☐ 时序图 ☐ 注意事项

-___-
32分钟前
1
0
分布式之数据库和缓存双写一致性方案解析

引言 为什么写这篇文章? 首先,缓存由于其高并发和高性能的特性,已经在项目中被广泛使用。在读取缓存方面,大家没啥疑问,都是按照下图的流程来进行业务操作。 但是在更新缓存方面,对于更...

别打我会飞
34分钟前
6
0
我的oracle11G,12c OCM之路

ocm认证感悟 ---------------------- 距离拿到ocm证书已经过了1年的时间,当初拿到证书的心情到现在还记得。其实在每个DBA心里都有一个成为强者的梦想,需要被认可,我也一样。我干过开发,做...

hnairdb
34分钟前
2
1
手动部署kubernetes集群(1.13.1最新版)

一、机器规划 使用五台机子部署k8s集群,规划如下: master节点3台(同时也是etcd节点) node节点2台 ip分配如下: ip:192.168.10.101,主机名:k8s-etcd01 ip:192.168.10.102,主机名:k8s...

人在艹木中
39分钟前
29
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部