Fork-Join分治编程总结(2)--类ForkJoinPool核心方法使用总结

￡神魔★判官ぃ 2022-02-24 14:28 214阅读 0赞

### 1.方法public void execute(ForkJoinTask<?> task)的使用 ###

在ForkJoinPool.java类中的execute()方法是以异步的方式执行任务

### 2.方法public void execute(Runnable task)的使用 ###

### 3.方法public void execute(ForkJoinTask<?> task)如何处理返回值 ###

虽然public void execute(ForkJoinTask<?> task)方法无返回值，但还是可以通过RecursiveTask对象处理返回值

package mytask;
    
    import java.util.concurrent.RecursiveTask;
    
    public class MyRecursiveTask extends RecursiveTask<String> {
    
    	@Override
    	protected String compute() {
    		try {
    			Thread.sleep(5000);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		return "我是返回值";
    	}
    }

package test;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import mytask.MyRecursiveTask;
    
    public class Run {
    
    	public static void main(String[] args) {
    		try {
    			MyRecursiveTask task = new MyRecursiveTask();
    			ForkJoinPool pool = new ForkJoinPool();
    			pool.execute(task);
    			// execute方法无返回值
    			// 想取得返回值得通过
    			// RecursiveTask对象
    			System.out.println(System.currentTimeMillis());
    			System.out.println(task.get());
    			System.out.println(System.currentTimeMillis());
    			Thread.sleep(5000);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		} catch (ExecutionException e) {
    			e.printStackTrace();
    		}
    	}
    }

### 4.方法public<T> ForkJoinTask<T> submit(ForkJoinTask<T> task)的使用 ###

方法execute()无返回值，submit()有返回值。

package test;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    import mytask.MyRecursiveTask;
    
    public class Run {
    
    	public static void main(String[] args) {
    		try {
    			MyRecursiveTask task = new MyRecursiveTask();
    			ForkJoinPool pool = new ForkJoinPool();
    			ForkJoinTask<String> returnTask = pool.submit(task);
    			System.out.println("begin " + System.currentTimeMillis());
    			System.out.println(returnTask.get());
    			System.out.println("  end " + System.currentTimeMillis());
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		} catch (ExecutionException e) {
    			e.printStackTrace();
    		}
    	}
    }

### 5.方法public ForkJoinTask<?> submit(Runnable task)的使用 ###

传入Runnable接口虽然没有返回值，但调用get()方法呈阻塞状态

package test;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    
    public class Run {
    
    	public static void main(String[] args) {
    		try {
    			ForkJoinPool pool = new ForkJoinPool();
    			System.out.println("begin " + System.currentTimeMillis());
    			ForkJoinTask task = pool.submit(new Runnable() {
    				@Override
    				public void run() {
    					try {
    						Thread.sleep(5000);
    						System.out.println("ThreadName=" + Thread.currentThread().getName());
    					} catch (InterruptedException e) {
    						e.printStackTrace();
    					}
    				}
    			});
    			System.out.println(task.get());
    			System.out.println("  end " + System.currentTimeMillis());
    			Thread.sleep(5000);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		} catch (ExecutionException e) {
    			e.printStackTrace();
    		}
    	}
    
    }

### 6.方法public<T> ForkJoinTask<T> submit(Callable<T> task)的使用 ###

package mycallable;
    
    import java.util.concurrent.Callable;
    
    public class MyCallable implements Callable<String> {
    
    	@Override
    	public String call() throws Exception {
    		try {
    			Thread.sleep(3000);
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    		return "我是返回值callableVersion";
    	}
    
    }

package test;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    import mycallable.MyCallable;
    
    public class Run {
    
    	public static void main(String[] args) {
    		try {
    			MyCallable callable = new MyCallable();
    
    			ForkJoinPool pool = new ForkJoinPool();
    			System.out.println("begin " + System.currentTimeMillis());
    			ForkJoinTask<String> returnTask = pool.submit(callable);
    			System.out.println(returnTask.get());
    			System.out.println("  end " + System.currentTimeMillis());
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		} catch (ExecutionException e) {
    			e.printStackTrace();
    		}
    	}
    }

### 7.方法public<T> ForkJoinTask<T> submit(Runnable task,T result)的使用 ###

因为该方法是异步运行的，submit方法运行的结果无法直接取到值。解决办法是加一个延时功能，可以加一个sleep()方法，但是无法确定Task任务执行完需要多少时间，所以该方法不是很方便，可以用future.get().getUsername()使用阻塞效果获得返回值。

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

package myrunnable;
    
    import entity.Userinfo;
    
    public class MyRunnable implements Runnable {
    
    	private Userinfo userinfo;
    
    	public MyRunnable(Userinfo userinfo) {
    		super();
    		this.userinfo = userinfo;
    	}
    
    	public void run() {
    		try {
    			userinfo.setUsername("设置的值");
    			System.out.println("已经设置完结！");
    			Thread.sleep(3000);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}
    }

package test.run;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.Future;
    
    import myrunnable.MyRunnable;
    import entity.Userinfo;
    
    public class Run {
    
    	public static void main(String[] args) throws InterruptedException, ExecutionException {
    		Userinfo userinfo = new Userinfo();
    		MyRunnable runnable = new MyRunnable(userinfo);
    
    		ForkJoinPool pool = new ForkJoinPool();
    		Future<Userinfo> future = pool.submit(runnable, userinfo);
    		System.out.println(future);
    		// 建议使用此种方式future.get()
    		// 因为get()方法呈阻塞效果
    		System.out.println("userinfo username=" + future.get().getUsername());
    	}
    }

### 8.方法public<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)的使用 ###

方法public<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)具有阻塞特性

package mycallable;
    
    import java.util.concurrent.Callable;
    
    public class MyCallable implements Callable<String> {
    
    	private long sleepValue;
    
    	public MyCallable(long sleepValue) {
    		super();
    		this.sleepValue = sleepValue;
    	}
    
    	@Override
    	public String call() throws Exception {
    		try {
    			System.out.println(Thread.currentThread().getName() + " sleep" + sleepValue + " nowTime："
    					+ System.currentTimeMillis());
    			Thread.sleep(sleepValue);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		return "我是返回值";
    	}
    
    }

package test;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.Future;
    
    import mycallable.MyCallable;
    
    public class Test {
    
    	public static void main(String[] args) {
    		try {
    			List list = new ArrayList();
    			list.add(new MyCallable(5000));
    			list.add(new MyCallable(4000));
    			list.add(new MyCallable(3000));
    			list.add(new MyCallable(2000));
    			list.add(new MyCallable(1000));
    
    			ForkJoinPool pool = new ForkJoinPool();
    			List<Future<String>> listFuture = pool.invokeAll(list);
    			for (int i = 0; i < listFuture.size(); i++) {
    				System.out.println(listFuture.get(i).get() + " nowTime：" + System.currentTimeMillis());
    			}
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		} catch (ExecutionException e) {
    			e.printStackTrace();
    		}
    	}
    }

### 9.方法public <T> T invoke(ForkJoinTask<T> task)的使用 ###

方法execute(task),submit(task)以及invoke(task)都可以在异步队列中执行任务，需要注意的是，方法invoke()是阻塞的，而它们在使用上的区别其实很简单，execute(task)只执行任务，没有返回值；而submit(task)方法有返回值，返回值类型是ForkJoinTask,想取得返回值时，需要使用ForkJoinTask对象的get()方法，而invoke(task)和submit(task)方法一样都具有返回值的功能，区别就是invoke(task)方法直接将返回值返回，而不是通过ForkJoinTask对象的get()方法。

package mytask;
    
    import java.util.concurrent.RecursiveTask;
    
    public class MyRecursiveTask extends RecursiveTask<String> {
    
    	@Override
    	protected String compute() {
    		return "我是返回值";
    	}
    }

package myaction;
    
    import java.util.concurrent.RecursiveAction;
    
    public class MyRecursiveAction extends RecursiveAction {
    
    	@Override
    	protected void compute() {
    		System.out.println("ThreadName=" + Thread.currentThread().getName());
    	}
    }

package test;
    
    import java.util.concurrent.ForkJoinPool;
    
    import myaction.MyRecursiveAction;
    
    public class Test1 {
    	public static void main(String[] args) throws InterruptedException {
    		MyRecursiveAction action = new MyRecursiveAction();
    		ForkJoinPool pool = new ForkJoinPool();
    		pool.invoke(action);
    	}
    }

package test;
    
    import java.util.concurrent.ForkJoinPool;
    
    import mytask.MyRecursiveTask;
    
    public class Test2 {
    	public static void main(String[] args) throws InterruptedException {
    		MyRecursiveTask action = new MyRecursiveTask();
    		ForkJoinPool pool = new ForkJoinPool();
    		String returnString = pool.invoke(action);
    		System.out.println(returnString);
    	}
    }

### 10.方法public void shutdown(),public void shutdownNow(),public List shutdownNow(),isTerminating(),isTerminated(),public boolean isShutdown()和public boolean awaitTermination(long timeout,TimeUnit unit)等方法的使用。 ###

这里就不总结了，用时再看书。