Java并发编程之CompletionService

简介：CompletionService功能是以异步的方式一边生成新任务，一边处理完成的任务，使得任务的执行与处理进行分离。使用submit提交任务，使用task获取已经完成的任务

先看个示例：

此功能是获取每个任务的执行结果，这里通过future来实现

public class FutureDemo {
	
	private static class CallableTask implements Callable<String> {

		private int time;
		private String name ;
		public CallableTask(int time, String name) {
			this.time = time ;
			this.name = name ;
		}
		
		@Override
		public String call() throws Exception {
			TimeUnit.SECONDS.sleep(this.time) ;
			return name ;
		}
		
	}
	
	public static void main(String[] args) throws Exception {
		ThreadPoolExecutor pool = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(20)) ;
		List<Callable<String>> tasks = new ArrayList<>() ;
		List<Future<String>> results = new ArrayList<>() ;
		for (int i = 1; i <= 5; i++) {
			tasks.add(new CallableTask(i, "name" + i)) ;
		}
		for (int i = 0; i < 5; i++) {
			results.add(pool.submit(tasks.get(i))) ;
		}
		for (int i = 4; i >= 0; i--) {
			System.out.println(results.get(i).get()) ;
		}
	}
	
}

执行结果：

这里提交了5个任务，分别等待1到5秒的时间，在获取任务执行结果的时候，不管哪个任务先执行都只会按照你的获取顺序去获取即便是有其它的任务已经先执行完。future具有阻塞功能。

接下来用CompletionService实现：

public class CompletionServiceDemo {
	
	private static class CallableTask implements Callable<String> {

		private int time;
		private String name ;
		public CallableTask(int time, String name) {
			this.time = time ;
			this.name = name ;
		}
		
		@Override
		public String call() throws Exception {
			TimeUnit.SECONDS.sleep(this.time) ;
			return name ;
		}
		
	}
	
	public static void main(String[] args) throws Exception {
		ThreadPoolExecutor pool = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(20)) ;
		List<Callable<String>> tasks = new ArrayList<>() ;
		CompletionService<String> cs = new ExecutorCompletionService<>(pool) ;
		for (int i = 1; i <= 5; i++) {
			tasks.add(new CallableTask(i, "name" + i)) ;
		}
		for (int i = 0; i < 5; i++) {
			cs.submit(tasks.get(i)) ;
		}
		for (int i = 4; i >= 0; i--) {
			System.out.println("--") ;
			System.out.println(cs.take().get()) ;
		}
	}
	
}

执行结果：

运行结果发现哪个任务先执行完take就会获取那个任务，当没有任务完成的时候take会阻塞。

CompletionService的poll方法非阻塞的，poll方法获取并删除已经完成的Future，如果没有就返回

null。

public class CompletionServiceDemo2 {
	
	private static class CallableTask implements Callable<String> {

		private int time;
		private String name ;
		public CallableTask(int time, String name) {
			this.time = time ;
			this.name = name ;
		}
		
		@Override
		public String call() throws Exception {
			TimeUnit.SECONDS.sleep(this.time) ;
			return name ;
		}
		
	}
	
	public static void main(String[] args) throws Exception {
		ThreadPoolExecutor pool = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(20)) ;
		CompletionService<String> cs = new ExecutorCompletionService<>(pool) ;
		for (int i = 0; i < 5; i++) {
			cs.submit(new CallableTask(i, "name" + i)) ;
		}
		for (int i = 4; i >= 0; i--) {
			System.out.println(cs.poll()) ;
		}
	}
	
}

执行结果：

获取到了第一个完成的任务，其它的都返回了null。

poll返回还可以设置超时时间

public static void main(String[] args) throws Exception {
		ThreadPoolExecutor pool = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(20)) ;
		CompletionService<String> cs = new ExecutorCompletionService<>(pool) ;
		for (int i = 0; i < 5; i++) {
			cs.submit(new CallableTask(i, "name" + i)) ;
		}
		for (int i = 4; i >= 0; i--) {
			System.out.println(cs.poll(5, TimeUnit.SECONDS)) ;
		}
	}

这里超时设置了5秒钟，如果在5秒内获取到了数据那就继续向下执行，如果超过了5秒也会继续向下执行。

CompletionService异常处理

示例：

public class CompletionServiceDemo3 {
	
	private static class CallableTask implements Callable<String> {

		private int time;
		private String name ;
		public CallableTask(int time, String name) {
			this.time = time ;
			this.name = name ;
		}
		
		@Override
		public String call() throws Exception {
			if (this.time == 2) {
				throw new RuntimeException("发生了异常：" + this.name) ;
			}
			TimeUnit.SECONDS.sleep(this.time) ;
			return name ;
		}
		
	}
	
	public static void main(String[] args) throws Exception {
		ThreadPoolExecutor pool = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(20)) ;
		CompletionService<String> cs = new ExecutorCompletionService<>(pool) ;
		for (int i = 0; i < 5; i++) {
			cs.submit(new CallableTask(i, "name" + i)) ;
		}
		for (int i = 4; i >= 0; i--) {
			System.out.println(cs.take()) ;
		}
	}
	
}

这里设置了当time为2时抛出异常，执行结果：

在这里获取结果的时候只是调用了take，并没有调用get方法获取值信息，所以这里跟没有异常一样没有反应出来。

获取返回值时：

public static void main(String[] args) throws Exception {
		ThreadPoolExecutor pool = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(20)) ;
		CompletionService<String> cs = new ExecutorCompletionService<>(pool) ;
		for (int i = 0; i < 5; i++) {
			cs.submit(new CallableTask(i, "name" + i)) ;
		}
		for (int i = 4; i >= 0; i--) {
			System.out.println(cs.take().get()) ;
		}
	}

执行结果：

当调用了cs.take().get()返回值时异常才抛出。从这也发现当有异常抛出时，其它还任务都将被终止运行。

CompletionService.submit(Runnable, V)指定返回值（当执行成功）

public class CompletionServiceDemo3 {
	
	private static class User {
		public String name ;
	}
	
	private static class Task implements Runnable {
		private User user ;
		public Task(User user) {
			this.user = user ;
		}
		@Override
		public void run() {
			try {
				TimeUnit.SECONDS.sleep(5) ;
			} catch (InterruptedException e) {
				e.printStackTrace();
			}			
			System.out.println("任务执行完成。。。") ;
			this.user.name = "张三" ;
		}
		
	}
	
	public static void main(String[] args) throws Exception {
		ThreadPoolExecutor pool = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(20)) ;
		CompletionService<User> cs = new ExecutorCompletionService<>(pool) ;
		User user = new User() ;
		Future<User> f = cs.submit(new Task(user), user) ;
		System.out.println(f.get().name) ;
	}
	
}

执行结果：