CompletionService使用总结

怼烎@ 2022-02-24 12:42 170阅读 0赞

## 接口CompletionService的功能是以异步的方式一边生产新的任务，一边处理已完成任务的结果，这样可以将执行任务与处理任务分离开来进行处理。使用submit执行任务，使用take取得已完成的任务，并按照完成这些任务的时间顺序处理它们的结果。 ##

### 1.CompletionService介绍 ###

接口CompletionService的结构比较简洁，仅有一个实现类ExecutorCompletionService，类ExecutorCompletionService需要依赖于Executor对象，大部分的实现也就是实现线程池ThreadPoolExecutor对象。

### 2.使用CompletionService解决Future的缺点 ###

### 3.使用take()方法 ###

方法take()取得最先完成任务的Future对象，谁执行时间最短谁最先返回，即方法take()是按任务执行的速度，从快到慢的顺序获得Future对象。

package test.run;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.CompletionService;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorCompletionService;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class Run {
    
    	public static void main(String[] args) {
    		try {
    			// take方法：获取并移除表示下一个已完成任务的 Future，如果目前不存在这样的任务，则等待。
    			ExecutorService executorService = Executors.newCachedThreadPool();
    			CompletionService csRef = new ExecutorCompletionService(executorService);
    			for (int i = 0; i < 10; i++) {
    				csRef.submit(new Callable<String>() {
    					public String call() throws Exception {
    						long sleepValue = (int) (Math.random() * 1000);
    						System.out.println("sleep=" + sleepValue + " " + Thread.currentThread().getName());
    						Thread.sleep(sleepValue);
    						return "haige：" + sleepValue + " " + Thread.currentThread().getName();
    					}
    				});
    			}
    			for (int i = 0; i < 10; i++) {
    				System.out.println(csRef.take().get());
    			}
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		} catch (ExecutionException e) {
    			e.printStackTrace();
    		}
    
    	}
    }

### 4.使用poll()方法 ###

方法poll()的作用是获取并移除表示下一个已完成任务的Future，如果不存在这样的任务，则返回null，方法poll()无阻塞的效果。

package test.run;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.CompletionService;
    import java.util.concurrent.ExecutorCompletionService;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class Run {
    
    	public static void main(String[] args) {
    		// poll方法：获取并移除表示下一个已完成任务的 Future，如果不存在这样的任务，则返回 null
    		ExecutorService executorService = Executors.newCachedThreadPool();
    		CompletionService csRef = new ExecutorCompletionService(executorService);
    		for (int i = 0; i < 1; i++) {
    			csRef.submit(new Callable<String>() {
    				public String call() throws Exception {
    					Thread.sleep(3000);
    					System.out.println("3秒过后了");
    					return "haige3s";
    				}
    			});
    		}
    		for (int i = 0; i < 1; i++) {
    			System.out.println(csRef.poll());
    		}
    	}
    }

### 5.使用poll(long timeout,TimeUnit unit) 方法 ###

方法Future<V> poll(long timeout,TimeUnit unit)的作用是等待指定的timeout时间，在timeout时间之内获取到值时立即向下继续执行，如果超时也立即向下执行。

package mycallable;
    
    import java.util.concurrent.Callable;
    
    public class MyCallableA implements Callable<String> {
    	public String call() throws Exception {
    		Thread.sleep(5000);
    		System.out.println("MyCallableA " + System.currentTimeMillis());
    		return "returnA";
    	}
    }

package mycallable;
    
    import java.util.concurrent.Callable;
    
    public class MyCallableB implements Callable<String> {
    	public String call() throws Exception {
    		Thread.sleep(10000);
    		System.out.println("MyCallableB " + System.currentTimeMillis());
    		return "returnB";
    	}
    }

package test.run;
    
    import java.util.concurrent.CompletionService;
    import java.util.concurrent.Executor;
    import java.util.concurrent.ExecutorCompletionService;
    import java.util.concurrent.Executors;
    
    import mycallable.MyCallableA;
    import mycallable.MyCallableB;
    
    /**
     * 一般状况poll()方法
     * 返回值两个null，因为任务未执行完毕
     * @author linhaiy
     * @date 2019.03.08
     */
    public class Run1 {
    
    	public static void main(String[] args) {
    
    		MyCallableA callableA = new MyCallableA();
    		MyCallableB callableB = new MyCallableB();
    
    		Executor executor = Executors.newCachedThreadPool();
    		CompletionService csRef = new ExecutorCompletionService(executor);
    		csRef.submit(callableA);
    		csRef.submit(callableB);
    
    		for (int i = 0; i < 2; i++) {
    			System.out.println("任务执行情况 ：" + " " + csRef.poll());
    		}
    		System.out.println("main end!");
    	}
    }

package test.run;
    
    import java.util.concurrent.CompletionService;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Executor;
    import java.util.concurrent.ExecutorCompletionService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    import mycallable.MyCallableA;
    import mycallable.MyCallableB;
    
    /**
     * 使用poll(long timeout,TimeUnit unit) 
     * 返回2个值，一共等待12秒
     * @author linhaiy
     *
     */
    public class Run2 {
    
    	public static void main(String[] args) {
    
    		try {
    			MyCallableA callableA = new MyCallableA();
    			MyCallableB callableB = new MyCallableB();
    
    			Executor executor = Executors.newCachedThreadPool();
    			CompletionService csRef = new ExecutorCompletionService(executor);
    			csRef.submit(callableA);
    			csRef.submit(callableB);
    
    			for (int i = 0; i < 2; i++) {
    				System.out.println("任务执行情况:" + " "
    						+ csRef.poll(6, TimeUnit.SECONDS).get());
    				System.out.println("X");
    			}
    			System.out.println("main end!");
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		} catch (ExecutionException e) {
    			e.printStackTrace();
    		}
    	}
    }

package test.run;
    
    import java.util.concurrent.CompletionService;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Executor;
    import java.util.concurrent.ExecutorCompletionService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    import mycallable.MyCallableA;
    import mycallable.MyCallableB;
    
    /**
     * 使用poll(long timeout,TimeUnit unit) 
     * 在第4秒时(任务执行情况1)打印的返回值是null,因为还未到达5秒，其他两次都正确得到额返回值，因为任务已经完成
     * @author linhaiy
     *
     */
    public class Run3 {
    
    	public static void main(String[] args) {
    
    		try {
    			MyCallableA callableA = new MyCallableA();
    			MyCallableB callableB = new MyCallableB();
    
    			Executor executor = Executors.newCachedThreadPool();
    			CompletionService csRef = new ExecutorCompletionService(executor);
    			csRef.submit(callableA);
    			csRef.submit(callableB);
    
    			System.out.println("任务执行情况1:" + " "
    					+ csRef.poll(4, TimeUnit.SECONDS) + " "
    					+ System.currentTimeMillis());
    			System.out.println("X");
    			System.out.println("任务执行情况2:" + " "
    					+ csRef.poll(2, TimeUnit.SECONDS).get() + " "
    					+ System.currentTimeMillis());
    			System.out.println("X");
    			System.out.println("任务执行情况3:" + " "
    					+ csRef.poll(5, TimeUnit.SECONDS).get() + " "
    					+ System.currentTimeMillis());
    			System.out.println("X");
    
    			System.out.println("main end!");
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		} catch (ExecutionException e) {
    			e.printStackTrace();
    		}
    	}
    }

package test.run;
    
    import java.util.concurrent.CompletionService;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Executor;
    import java.util.concurrent.ExecutorCompletionService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    import mycallable.MyCallableA;
    import mycallable.MyCallableB;
    
    public class Run4 {
    
    	public static void main(String[] args) {
    
    		try {
    			MyCallableA callableA = new MyCallableA();
    			MyCallableB callableB = new MyCallableB();
    
    			Executor executor = Executors.newCachedThreadPool();
    			CompletionService csRef = new ExecutorCompletionService(executor);
    			csRef.submit(callableA);
    			csRef.submit(callableB);
    
    			System.out.println("任务执行情况1:" + " "
    					+ csRef.poll(4, TimeUnit.SECONDS) + " "
    					+ System.currentTimeMillis());
    			System.out.println("X");
    			System.out.println("任务执行情况2:" + " "
    					+ csRef.poll(7, TimeUnit.SECONDS).get() + " "
    					+ System.currentTimeMillis());
    			System.out.println("X");
    
    			System.out.println("main end!");
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		} catch (ExecutionException e) {
    			e.printStackTrace();
    		}
    	}
    }

说明了方法poll()中的timeout的值如果小于任务执行的时间，则返回值就是null。

### 6.方法Future<V> submit(Runnable task,V result) 的测试 ###

参数V是submit()方法的返回值

package entity;
    
    public class Userinfo {
    
    	private String username;
    	private String password;
    
    	public Userinfo() {
    		super();
    	}
    
    	public Userinfo(String username, String password) {
    		super();
    		this.username = username;
    		this.password = password;
    	}
    
    	public String getUsername() {
    		return username;
    	}
    
    	public void setUsername(String username) {
    		this.username = username;
    	}
    
    	public String getPassword() {
    		return password;
    	}
    
    	public void setPassword(String password) {
    		this.password = password;
    	}
    }

package myrunnable;
    
    import entity.Userinfo;
    
    public class MyRunnable implements Runnable {
    	private Userinfo userinfo;
    
    	public MyRunnable(Userinfo userinfo) {
    		super();
    		this.userinfo = userinfo;
    	}
    
    	@Override
    	public void run() {
    		userinfo.setUsername("usernameValue");
    		userinfo.setPassword("passwordValue");
    		System.out.println("run 运行了 ");
    	}
    }

### 总结：接口CompletionService完全可以避免FutureTask类阻塞的缺点，可更加有效地处理Future的返回值，也就是哪个任务先执行完，CompletionService就先取得这个任务的返回值再处理。 ###