Java线程之fork/join框架

Myth丶恋晨 2021-09-12 05:32 258阅读 0赞

fork/join框架是用多线程的方式实现分治法来解决问题。fork指的是将问题不断地缩小规模，join是指根据子问题的计算结果，得出更高层次的结果。

fork/join框架的使用有一定的约束条件：

1. 除了fork()  和  join()方法外，线程不得使用其他的同步工具。线程最好也不要sleep()

2. 线程不得进行I/O操作

3. 线程不得抛出checked exception

此框架有几个核心类：ForkJoinPool是实现了工作窃取[算法][Link 1]的线程池。ForkJoinTask是任务类，他有2个子类：RecursiveAction无返回值，RecursiveTask有返回值，在定义自己的任务时，一般都是从这2类中挑一个，通过继承的方式定义自己的新类。由于ForkJoinTask类实现了Serializable接口，因此，定义自己的任务类时，应该定义serialVersionUID属性。

在编写任务时，推荐的写法是这样的：

**\[java\]**  [view plain][]   [copy][view plain]

1.  If (problem size > **default** size)\{  
2.  task s = divide(task);  
3.  execute(tasks);  
4.  \} **else** \{  
5.  resolve problem using another algorithm;  
6.  \}

ForkJoinPool实现了工作窃取算法（work-stealing），线程会主动寻找新创建的任务去执行，从而保证较高的线程利用率。它使用守护线程（deamon）来执行任务，因此无需对他显示的调用shutdown()来关闭。一般情况下，一个程序只需要唯一的一个ForkJoinPool，因此应该按如下方式创建它：

static final ForkJoinPool mainPool = new ForkJoinPool(); //线程的数目等于CPU的核心数

下面给出一个非常简单的例子，功能是将一个数组中每一个元素的值加1。具体实现为：将大数组不断分解为更短小的子数组，当子数组长度不超过10的时候，对其中所有元素进行加1操作。

**\[java\]**  [view plain][]   [copy][view plain]

1.  **public** **class** Test \{  
2.        
3.      **public** **final** **static** ForkJoinPool mainPool = **new** ForkJoinPool();  
4.        
5.      **public** **static** **void** main(String\[\] args)\{  
6.          **int** n = 26;  
7.          **int**\[\] a = **new** **int**\[n\];  
8.          **for**(**int** i=0; i<n; i++) \{  
9.              a\[i\] = i;  
10.         \}  
11.         SubTask task = **new** SubTask(a, 0, n);  
12.         mainPool.invoke(task);  
13.         **for**(**int** i=0; i<n; i++) \{  
14.             System.out.print(a\[i\]+" ");  
15.         \}  
16.     \}  
17. \}  
18.   
19. **class** SubTask **extends** RecursiveAction \{  
20.   
21.     **private** **static** **final** **long** serialVersionUID = 1L;  
22.       
23.     **private** **int**\[\] a;  
24.     **private** **int** beg;  
25.     **private** **int** end;  
26.       
27.     **public** SubTask(**int**\[\] a, **int** beg, **int** end) \{  
28.         **super**();  
29.         **this**.a = a;  
30.         **this**.beg = beg;  
31.         **this**.end = end;  
32.     \}  
33.   
34.     @Override  
35.     **protected** **void** compute() \{  
36.         **if**(end-beg>10) \{  
37.             **int** mid = (beg+end) / 2;  
38.             SubTask t1 = **new** SubTask(a, beg, mid);  
39.             SubTask t2 = **new** SubTask(a, mid, end);  
40.             invokeAll(t1, t2);  
41.         \}**else** \{  
42.             **for**(**int** i=beg; i<end; i++) \{  
43.                 a\[i\] = a\[i\] + 1;  
44.             \}  
45.         \}  
46.     \}  
47. \}

例子2，任务拥有返回值。随机生成一个数组，每个元素均是0-999之间的整数，统计该数组中每个数字出现1的次数的和。

实现方法，将该数组不断的分成更小的数组，直到每个子数组的长度为1，即只包含一个元素。此时，统计该元素中包含1的个数。最后汇总，得到数组中每个数字共包含了多少个1。

**\[java\]**  [view plain][]   [copy][view plain]

1.  **public** **class** Test \{  
2.        
3.      **public** **final** **static** ForkJoinPool mainPool = **new** ForkJoinPool();  
4.        
5.      **public** **static** **void** main(String\[\] args)\{  
6.          **int** n = 26;  
7.          **int**\[\] a = **new** **int**\[n\];  
8.          Random rand = **new** Random();  
9.          **for**(**int** i=0; i<n; i++) \{  
10.             a\[i\] = rand.nextInt(1000);  
11.         \}  
12.         SubTask task = **new** SubTask(a, 0, n);  
13.         **int** count = mainPool.invoke(task);  
14.         **for**(**int** i=0; i<n; i++) \{  
15.             System.out.print(a\[i\]+" ");  
16.         \}  
17.         System.out.println("\\n数组中共出现了" + count + "个1");  
18.     \}  
19. \}  
20.   
21. **class** SubTask **extends** RecursiveTask<Integer> \{  
22.   
23.     **private** **static** **final** **long** serialVersionUID = 1L;  
24.       
25.     **private** **int**\[\] a;  
26.     **private** **int** beg;  
27.     **private** **int** end;  
28.       
29.     **public** SubTask(**int**\[\] a, **int** beg, **int** end) \{  
30.         **super**();  
31.         **this**.a = a;  
32.         **this**.beg = beg;  
33.         **this**.end = end;  
34.     \}  
35.   
36.     @Override  
37.     **protected** Integer compute() \{  
38.         **int** result = 0;  
39.         **if**(end-beg>1) \{  
40.             **int** mid = (beg+end)/2;  
41.             SubTask t1 = **new** SubTask(a, beg, mid);  
42.             SubTask t2 = **new** SubTask(a, mid, end);  
43.             invokeAll(t1, t2);  
44.             **try** \{  
45.                 result = t1.get()+t2.get();  
46.             \} **catch** (InterruptedException | ExecutionException e) \{  
47.                 e.printStackTrace();  
48.             \}  
49.         \} **else** \{  
50.             result = count(a\[beg\]);  
51.         \}  
52.         **return** result;  
53.     \}  
54.       
55.     //统计一个整数中出现了几个1  
56.     **private** **int** count(**int** n) \{  
57.         **int** result = 0;  
58.         **while**(n>0) \{  
59.             **if**(n % 10==1) \{  
60.                 result++;  
61.             \}  
62.             n = n / 10;  
63.         \}  
64.         **return** result;  
65.     \}  
66. \}

例子3，异步执行任务。前面两个例子都是同步执行任务，当启动任务后，主线程陷入了阻塞状态，直到任务执行完毕。若创建新任务后，希望当前线程能继续执行而非陷入阻塞，则需要异步执行。ForkJoinPool线程池提供了execute()方法来异步启动任务，而作为任务本身，可以调用fork()方法异步启动新的子任务，并调用子任务的join()方法来取得计算结果。需要注意的是，异步使用ForkJoin框架，无法使用“工作窃取”算法来提高线程的利用率，针对每个子任务，系统都会启动一个新的线程。

本例的功能是查找硬盘上某一类型的文件。给定文件扩展名后，将硬盘上所有该类型的文件名打印显示出来。作为主程序，启动任务后，继续显示任务的执行进度，每3秒钟打印显示一个黑点，表示任务在继续。最后，当所有线程都结束了，打印显示结果。

**\[java\]**  [view plain][]   [copy][view plain]

1.  **public** **class** ThreadLocalTest \{  
2.    
3.      **public** **static** **void** main(String\[\] args) **throws** Exception \{  
4.          Path p = Paths.get("D:/");  
5.          List<Path> roots = (List<Path>) FileSystems.getDefault().getRootDirectories();  
6.          List<Path> result = **new** ArrayList<>();  
7.          List<MyTask> tasks = **new** ArrayList<>();  
8.          ForkJoinPool pool = **new** ForkJoinPool();  
9.          **for**(Path root:roots) \{  
10.             MyTask t = **new** MyTask(root, "pdf");  
11.             pool.execute(t);  
12.             tasks.add(t);  
13.         \}  
14.           
15.         System.out.print("正在处理中");  
16.         **while**(isAllDone(tasks) == **false**) \{  
17.             System.out.print(". ");  
18.             TimeUnit.SECONDS.sleep(3);  
19.         \}  
20.           
21.         **for**(MyTask t:tasks) \{  
22.             result.addAll(t.get());  
23.         \}  
24.           
25.         **for**(Path pp:result) \{  
26.             System.out.println(pp);  
27.         \}  
28.     \}  
29.       
30.     **private** **static** **boolean** isAllDone(List<MyTask> tasks) \{  
31.         **boolean** result = **true**;  
32.         **for**(MyTask t:tasks) \{  
33.             **if**(t.isDone() == **false**) \{  
34.                 result = **false**;  
35.                 **break**;  
36.             \}  
37.         \}  
38.         **return** result;  
39.     \}  
40. \}  
41.   
42. **class** MyTask **extends** RecursiveTask<List<Path>> \{  
43.   
44.     **private** **static** **final** **long** serialVersionUID = 1L;  
45.       
46.     **private** Path path;  
47.     **private** String fileExtention;  
48.   
49.     **public** MyTask(Path path, String fileExtention) \{  
50.         **super**();  
51.         **this**.path = path;  
52.         **this**.fileExtention = fileExtention;  
53.     \}  
54.   
55.     @Override  
56.     **protected** List<Path> compute() \{  
57.         List<Path> result = **new** ArrayList<>();  
58.         **try** \{  
59.             DirectoryStream<Path> paths = Files.newDirectoryStream(path);  
60.             List<MyTask> subTasks = **new** ArrayList<>();  
61.             **for**(Path p:paths) \{  
62.                 **if**(Files.isDirectory(p)) \{  
63.                     MyTask t = **new** MyTask(p, fileExtention);  
64.                     t.fork();  
65.                     subTasks.add(t);  
66.                 \}**else** **if**(Files.isRegularFile(p)) \{  
67.                     **if**(p.toString().toLowerCase().endsWith("."\+fileExtention)) \{  
68.                         result.add(p);  
69.                     \}  
70.                 \}  
71.             \}  
72.               
73.             **for**(MyTask t:subTasks) \{  
74.                 result.addAll(t.join());  
75.             \}  
76.         \} **catch** (IOException e) \{  
77.         \}  
78.         **return** result;  
79.     \}  
80. \}

[Link 1]: http://lib.csdn.net/base/datastructure
[view plain]: http://blog.csdn.net/andycpp/article/details/8903155#