并发编程 ForkJoin框架

Author: xiaolyuh123

spring-boot-student-concurrent/src/main/java/com/xiaolyuh/ForkJoinCountTask.java

@@ -0,0 +1,74 @@
+package com.xiaolyuh;
+
+import java.util.Objects;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.RecursiveTask;
+
+/**
+ * 计算1+2+3+...+n的值
+ * 使用同步执行的方式
+ *
+ * @author yuhao.wang3
+ * @since 2019/6/25 17:07
+ */
+public class ForkJoinCountTask extends RecursiveTask<Long> {
+
+    /**
+     * 阀值
+     */
+    private int threshold = 10;
+
+    /**
+     * 任务的开始值
+     */
+    private long start;
+
+    /**
+     * 任务的结束值
+     */
+    private long end;
+
+    public ForkJoinCountTask(long start, long end) {
+        this.start = start;
+        this.end = end;
+    }
+
+    @Override
+    protected Long compute() {
+        if (end - start <= threshold) {
+            long count = 0;
+            for (int i = 0; i <= end - start; i++) {
+                count = count + start + i;
+            }
+            return count;
+        } else {
+            // 如果任务大于阈值，就分裂成三个子任务计算
+            long slip = (end - start) / 3;
+            ForkJoinCountTask oneTask = new ForkJoinCountTask(start, start + slip);
+            ForkJoinCountTask twoTask = new ForkJoinCountTask(start + slip + 1, start + slip * 2);
+            ForkJoinCountTask threeTask = new ForkJoinCountTask(start + slip * 2 + 1, end);
+            // 提交子任务到框架去执行
+            invokeAll(oneTask, twoTask, threeTask);
+            // 等待子任务执行完，得到其结果，并合并子任务
+            return oneTask.join() + twoTask.join() + threeTask.join();
+        }
+    }
+
+    public static void main(String[] args) {
+        long start = System.currentTimeMillis();
+        ForkJoinPool pool = new ForkJoinPool();
+        // 生成一个计算任务，负责计算1+2+3+n
+        ForkJoinCountTask countTask = new ForkJoinCountTask(1, 1000000);
+        // 执行一个任务（同步执行，任务会阻塞在这里直到任务执行完成）
+        pool.invoke(countTask);
+        // 异常检查
+        if (countTask.isCompletedAbnormally()) {
+            Throwable throwable = countTask.getException();
+            if (Objects.nonNull(throwable)) {
+                System.out.println(throwable.getMessage());
+            }
+        }
+        // join方法是一个阻塞方法，会等待任务执行完成
+        System.out.println("计算为：" + countTask.join() + ", 耗时：" + (System.currentTimeMillis() - start) + "毫秒");
+    }
+}

spring-boot-student-concurrent/src/main/java/com/xiaolyuh/ForkJoinSearchFileTask.java

@@ -0,0 +1,82 @@
+package com.xiaolyuh;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.RecursiveAction;
+
+/**
+ * 搜索指定目录下的指定文件
+ * 使用异步执行的方式
+ *
+ * @author yuhao.wang3
+ * @since 2019/6/25 17:07
+ */
+public class ForkJoinSearchFileTask extends RecursiveAction {
+
+    /**
+     * 指定目录
+     */
+    private File file;
+
+    /**
+     * 文件后缀
+     */
+    private String suffix;
+
+    public ForkJoinSearchFileTask(File file, String suffix) {
+        this.file = file;
+        this.suffix = suffix;
+    }
+
+    @Override
+    protected void compute() {
+        if (Objects.isNull(file)) {
+            return;
+        }
+
+        File[] files = file.listFiles();
+        List<ForkJoinSearchFileTask> fileTasks = new ArrayList<>();
+        if (Objects.nonNull(files)) {
+            for (File f : files) {
+                // 拆分任务
+                if (f.isDirectory()) {
+                    fileTasks.add(new ForkJoinSearchFileTask(f, suffix));
+                } else {
+                    if (f.getAbsolutePath().endsWith(suffix)) {
+                        System.out.println("文件： " + f.getAbsolutePath());
+                    }
+                }
+            }
+            // 提交并执行任务
+            invokeAll(fileTasks);
+            for (ForkJoinSearchFileTask fileTask : fileTasks) {
+                // 等待任务执行完成
+                fileTask.join();
+            }
+        }
+    }
+
+    public static void main(String[] args) throws Exception {
+        File file = new File("d:/");
+        ForkJoinPool pool = new ForkJoinPool();
+        // 生成一个计算任务，负责查找指定木目录
+        ForkJoinSearchFileTask searchFileTask = new ForkJoinSearchFileTask(file, ".txt");
+        // 异步执行一个任务
+        pool.execute(searchFileTask);
+
+        Thread.sleep(10);
+
+        // 做另外的事情
+        int count = 0;
+        for (int i = 0; i < 1000; i++) {
+            count += i;
+        }
+        System.out.println("计算任务：" + count);
+
+        // join方法是一个阻塞方法，会等待任务执行完成
+        searchFileTask.join();
+    }
+}

spring-boot-student-concurrent/src/main/java/com/xiaolyuh/JoinThread.java

@@ -0,0 +1,26 @@
+package com.xiaolyuh;
+
+/**
+ * 守护线程
+ *
+ * @author yuhao.wang3
+ */
+public class JoinThread {
+
+    public static void main(String[] args) throws InterruptedException {
+        Thread thread1 = new Thread(() -> System.out.println("线程1 " + Thread.currentThread().getName() + "执行完了"));
+
+        Thread thread2 = new Thread(() -> {
+            System.out.println("线程2 " + Thread.currentThread().getName() + " 开始执行");
+            try {
+                thread1.join();
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+            System.out.println("线程2 " + Thread.currentThread().getName() + "执行完了");
+        });
+//        thread.setDaemon(true);
+//        thread.start();
+        Thread.sleep(500);
+    }
+}