thread_fork/join并发框架2

转自  http://blog.csdn.net/mr_zhuqiang/article/details/48300229

三.使用异步方式

　　invokeAll(task1,task2);　是同步方式，将当前任务挂起直到子任务发送到 Fork/join线程池中执行完成，这种方式允许工作窃取算法，分配一个新任务给在执行休眠任务的工作者线程。
　　相反当采用异步的方式（比如，fork()），任务将继续执行，所以就没有办法使用工作窃取算法了。因为不存在等待的线程了。除非使用join() 或则 get() 来获取结果，等待任务的完成。
　　invokeAll()采用同步方式，工作者线程将会休眠等待子任务的完成，所以能使用窃取算法派给工作者一个新任务
　　fork() 采用异步方式，只有结合join()或者get() 来等待任务的完成，进而可以使用窃取算法来提高性能。
　　get():如果ForkJoinTask类执行结束，或则一直等到结束，那么get()方法的这个版本则返回由compute()方法返回的结果
　　get() 方法 和 join()方法的区别：
　　join()方法不能被中断，如果中断join()方法的线程，方法将抛出Interrupted异常
　　如果任务抛出任何运行时异常，那么get()方法将返回ExecutionException异常，但是join方法返回的是RuntimeException；

public class ForkJoin3Test {
    public static void main(String[] args) throws InterruptedException {
        ForkJoinPool pool = new ForkJoinPool();
        Task3 mp3 = new Task3("C:\360CloudUI", "mp3");
        pool.execute(mp3);

        do {
            System.out.println("***********   状态信息巡查 ***************");
            System.out.printf("最大并行任务:%s,当前活动任务数(不准确的):%s,队列中的任务数量:%s,窃取数量:%s
", 
                    pool.getParallelism(),
                    pool.getActiveThreadCount(), 
                    pool.getQueuedTaskCount(), 
                    pool.getStealCount());
            TimeUnit.MILLISECONDS.sleep(10);
        } while (!mp3.isDone()); // 未完成则一直循环获取状态信息

        pool.shutdown();
        List<String> join = mp3.join();
        System.out.println("共找到符合的文件数量：" + join.size());
        for (String s : join) {
            System.out.println(s);
        }
    }
}

class Task3 extends RecursiveTask<List<String>> {
    private static final long serialVersionUID = 1L;
    private String path; // 文件夹路径
    private String suffix; // 后缀

    public Task3(String path, String suffix) {
        this.path = path;
        this.suffix = suffix;
    }
    @Override
    protected List<String> compute() {
        List<String> result = new ArrayList<String>(); // 存储结果
        List<Task3> tasks = new ArrayList<Task3>();    // 存储任务

        File file = new File(path);
        File[] files = file.listFiles();
        for (File f : files) {     // 分发和执行任务
            if (f.isDirectory()) { // 如果是文件夹，则使用异步的方式发送一个任务去执行
                Task3 task = new Task3(f.getAbsolutePath(), suffix);
                task.fork();      // 拆分任务异步执行
                tasks.add(task);
            } else {
                String name = f.getName();
                if (name.endsWith(suffix)) {
                    result.add(name);
                }
            }
        }
        if (tasks.size() > 1) {    // 如果当前任务大于1个 则打印信息
            System.out.printf("%s,tasks size(当前路径有) = %s个（文件夹），当前路径是：%s
", Thread.currentThread().getName(),
                    tasks.size(), path);
        }
        for (Task3 task : tasks) {           // 获取当前任务的结果
            List<String> join = task.join(); // 调用join方法等待任务完成
            result.addAll(join);             // 把任务结果添加到当前任务的结果中
        }
        return result;
    }
}

四 取消任务

  ForkJoinTask　对象中有一个cancel()方法来取消未开始的任务。取消任务有以下两点需要注意：
   1. ForkJoinPool类不提供任何方法来取消线程池中正在运行或则等待运行的所有任务。
   2. 取消任务时，不能取消已经被执行的任务。

public class ForkJoin4Test {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        int[] arrs = new ArrayGenerator().generateArray(50);
        //
        TaskManger taskManger = new TaskManger();
        ForkJoinPool pool = new ForkJoinPool();
        SearchNumberTask task = new SearchNumberTask(arrs, 0, arrs.length, 50, taskManger);

        pool.execute(task);
        pool.shutdown();
        pool.awaitTermination(1, TimeUnit.MILLISECONDS);

        // System.out.println("main:结束：" + task.get());
    }
}

// 数组生成
class ArrayGenerator {
    public int[] generateArray(int size) {
        int[] array = new int[size];
        Random random = new Random();
        for (int i = 0; i < size; i++) {
            array[i] = random.nextInt(10);
        }
        return array;
    }
}

// 任务管理类
class TaskManger {
    private List<ForkJoinTask<Integer>> tasks = new ArrayList<ForkJoinTask<Integer>>();

    public void addTask(ForkJoinTask<Integer> task) {
        tasks.add(task);
    }

    public void cancelTasks(ForkJoinTask<Integer> cancelTask) {
        for (ForkJoinTask<Integer> task : tasks) {
            if (task != cancelTask) {
                task.cancel(true);
                ((SearchNumberTask) task).writeCanceMesg();
            }
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

class SearchNumberTask extends RecursiveTask<Integer> {
    private int[] numbers;
    private int start, end;
    private int number;
    private TaskManger taskManger;

    private final static int NOT_FOUND = -1;

    public SearchNumberTask(int[] numbers, int start, int end, int number, TaskManger taskManger) {
        this.numbers = numbers;
        this.start = start;
        this.end = end;
        this.number = number;
        this.taskManger = taskManger;
    }

    @Override
    protected Integer compute() {

        int ret;
        if (end - start > 10) { // 拆分任务

            ret = launchTasks();

        } else { // 执行查找
            System.out.println("Task:开始：" + start + ":" + end);
            ret = lookForNumber();
            System.out.println("Task:结束--------：" + start + ":" + end);
        }

        return ret;
    }

    /**
     * 查找数字
     * 
     * @return
     */
    private int lookForNumber() {
        for (int i = start; i < end; i++) {
            if (numbers[i] == number) {
                System.out.printf("Task：目标number:%s已被找到,索引位置：%s
", number, i);
                taskManger.cancelTasks(this);
                return i;
            }
        }
        return NOT_FOUND;
    }

    /**
     * 拆分任务
     * 
     * @return
     */
    private int launchTasks() {
        int mid = (start + end) / 2;
        SearchNumberTask task1 = new SearchNumberTask(numbers, start, mid, number, taskManger);
        SearchNumberTask task2 = new SearchNumberTask(numbers, mid, end, number, taskManger);
        taskManger.addTask(task1);
        taskManger.addTask(task2);
        task1.fork(); // 异步执行
        task2.fork();

        int result = task1.join();
        if (result != -1) {
            return result;
        }
        return task2.join();
    }

    /** 取消任务 信息 **/
    public void writeCanceMesg() {
        System.out.printf("Task:取消了，start=%s,end=%s
", start, end);
    }
}

五 运行异常

Java有两种类型的异常：
　　非运行时异常（Checked Exception）：必须在方法上通过throws 子句抛出，或则通过try…catch语句进行扑捉处理。
　　运行时异常（Unchecked Exception）:不是强制的需要捕捉处理和throws抛出。
在ForkJoinTask类的compute方法中不能抛出非运行时异常，因为该方法没有throws的声明，根据Java重新方法的规则，所以不能抛出。而且在该compute中抛出的运行时异常，给我最明显直观的结果是，只要不调用get()获取结果，控制台是不会打印异常信息的。也就是说，异常被吞噬了。但是我们可以通过该类的其他方法来获取该异常。
　　task.isCompletedNormally() ： 任务完成时没有出错
　　task.isCompletedAbnormally() ： 来检查任务是否已经抛出异常或已经被取消了，要注意此方法。由于提交任务之后，检测该任务是否有异常，不是阻塞的。所以需要等待任务的完成。才能正确的获取到是否有异常
　　task.getException() : 获得任务中抛出的异常
该类中抛出的异常，只要一抛出异常，子任务都不会再继续执行。（反正就是说只要抛出了异常，任务结果肯定是不正确的了）
　　completeExceptionally(Throwable ex) ： 该方法 可以在语义上抛出一个异常，包括非运行时异常。要在获取结果前 通过task.isCompletedAbnormally()来配合操作。

public class ForkJoin5Test {
    public static void main(String[] args) throws InterruptedException {
        int[] arrs = { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };
        ForkJoinPool pool = new ForkJoinPool(1);
        Task5 task = new Task5(arrs, 0, arrs.length);
        pool.execute(task);

        pool.shutdown(); // 关闭执行器，并配合超时。来等待任务运行完成,发现一个特性：该执行器如来里面没有可活动的任务。执行器会自动关闭。而且调用get会阻塞任务直到返回结果
        pool.awaitTermination(1, TimeUnit.DAYS);

        // task.isCompletedNormally() 任务完成时没有出错
        if (task.isCompletedAbnormally()) { // 来检查任务是否已经抛出异常或已经被取消了，要注意此方法。由于提交任务之后，检测该任务是否有异常，不是阻塞的。所以需要上面的等待任务的完成。才能正确的获取到是否有异常
            System.out.println("检测到任务中有抛出的异常：" + task.getException().getMessage());
        } else {
            System.out.println(task.join());
        }
    }
}

class Task5 extends RecursiveTask<Integer> {
    private int[] arrs; // 要处理的数据
    private int start; // 开始索引
    private int end; // 结束索引

    public Task5(int[] arrs, int start, int end) {
        this.arrs = arrs;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        int result = 0;
        if (end - start < 2) {
            for (int i = start; i < end; i++) {
                result += arrs[i];
            }
            System.out.printf("%s,结果：%s
", Thread.currentThread().getName(), result);
            return result;
        } else {
            int mid = (start + end) / 2;
            // System.out.println(mid);
            if (mid == 2) {
                throw new RuntimeException("故意抛出的测试异常"); // 为了测试抛出异常，可以
                                                            // 关闭测异常。运行查看结果
                // Exception e = new Exception("故意抛出的非运行时异常");
                // completeExceptionally(e); //也可以使用 该方法，设置一个异常，因为 源码
                // setExceptionalCompletion
                // 是设置的异常，就相当于该异常并没有被抛出。在语义上通过task.isCompletedAbnormally()来抛出了非运行时异常
                // return null; // 如果不返回，程序将继续执行后面的代码，并不能达到真正抛出异常的效果
            }
            // 拆分成2个子任务继续检测和执行
            Task5 task1 = new Task5(arrs, start, mid);
            Task5 task2 = new Task5(arrs, mid, end);
            invokeAll(task1, task2); // 使用同步的方式 执行
            try {
                result = task1.get() + task2.get(); // 把子任务返回的结果相加
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        return result;
    }
}