av一区二区在线观看_亚洲男人的天堂网站_日韩亚洲视频_在线成人免费_欧美日韩精品免费观看视频_久草视

您的位置:首頁技術(shù)文章
文章詳情頁

java中的forkjoin框架的使用

瀏覽:6日期:2022-09-04 10:14:40

fork join框架是java 7中引入框架,這個框架的引入主要是為了提升并行計算的能力。

fork join主要有兩個步驟,第一就是fork,將一個大任務(wù)分成很多個小任務(wù),第二就是join,將第一個任務(wù)的結(jié)果join起來,生成最后的結(jié)果。如果第一步中并沒有任何返回值,join將會等到所有的小任務(wù)都結(jié)束。

還記得之前的文章我們講到了thread pool的基本結(jié)構(gòu)嗎?

ExecutorService - ForkJoinPool 用來調(diào)用任務(wù)執(zhí)行。 workerThread - ForkJoinWorkerThread 工作線程,用來執(zhí)行具體的任務(wù)。 task - ForkJoinTask 用來定義要執(zhí)行的任務(wù)。

下面我們從這三個方面來詳細(xì)講解fork join框架。

ForkJoinPool

ForkJoinPool是一個ExecutorService的一個實(shí)現(xiàn),它提供了對工作線程和線程池的一些便利管理方法。

public class ForkJoinPool extends AbstractExecutorService

一個work thread一次只能處理一個任務(wù),但是ForkJoinPool并不會為每個任務(wù)都創(chuàng)建一個單獨(dú)的線程,它會使用一個特殊的數(shù)據(jù)結(jié)構(gòu)double-ended queue來存儲任務(wù)。這樣的結(jié)構(gòu)可以方便的進(jìn)行工作竊取(work-stealing)。

什么是work-stealing呢?

默認(rèn)情況下,work thread從分配給自己的那個隊列頭中取出任務(wù)。如果這個隊列是空的,那么這個work thread會從其他的任務(wù)隊列尾部取出任務(wù)來執(zhí)行,或者從全局隊列中取出。這樣的設(shè)計可以充分利用work thread的性能,提升并發(fā)能力。

下面看下怎么創(chuàng)建一個ForkJoinPool。

最常見的方法就是使用ForkJoinPool.commonPool()來創(chuàng)建,commonPool()為所有的ForkJoinTask提供了一個公共默認(rèn)的線程池。

ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();

另外一種方式是使用構(gòu)造函數(shù):

ForkJoinPool forkJoinPool = new ForkJoinPool(2);

這里的參數(shù)是并行級別,2指的是線程池將會使用2個處理器核心。

ForkJoinWorkerThread

ForkJoinWorkerThread是使用在ForkJoinPool的工作線程。

public class ForkJoinWorkerThread extends Thread}

和一般的線程不一樣的是它定義了兩個變量:

final ForkJoinPool pool; // the pool this thread works in final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics

一個是該worker thread所屬的ForkJoinPool。 另外一個是支持 work-stealing機(jī)制的Queue。

再看一下它的run方法:

public void run() { if (workQueue.array == null) { // only run once Throwable exception = null; try { onStart(); pool.runWorker(workQueue); } catch (Throwable ex) { exception = ex; } finally { try { onTermination(exception); } catch (Throwable ex) { if (exception == null) exception = ex; } finally { pool.deregisterWorker(this, exception); } } } }

簡單點(diǎn)講就是從Queue中取出任務(wù)執(zhí)行。

ForkJoinTask

ForkJoinTask是ForkJoinPool中運(yùn)行的任務(wù)類型。通常我們會用到它的兩個子類:RecursiveAction和RecursiveTask<V>。

他們都定義了一個需要實(shí)現(xiàn)的compute()方法用來實(shí)現(xiàn)具體的業(yè)務(wù)邏輯。不同的是RecursiveAction只是用來執(zhí)行任務(wù),而RecursiveTask<V>可以有返回值。

既然兩個類都帶了Recursive,那么具體的實(shí)現(xiàn)邏輯也會跟遞歸有關(guān),我們舉個使用RecursiveAction來打印字符串的例子:

public class CustomRecursiveAction extends RecursiveAction { private String workload = ''; private static final int THRESHOLD = 4; private static Logger logger = Logger.getAnonymousLogger(); public CustomRecursiveAction(String workload) { this.workload = workload; } @Override protected void compute() { if (workload.length() > THRESHOLD) { ForkJoinTask.invokeAll(createSubtasks()); } else { processing(workload); } } private List<CustomRecursiveAction> createSubtasks() { List<CustomRecursiveAction> subtasks = new ArrayList<>(); String partOne = workload.substring(0, workload.length() / 2); String partTwo = workload.substring(workload.length() / 2, workload.length()); subtasks.add(new CustomRecursiveAction(partOne)); subtasks.add(new CustomRecursiveAction(partTwo)); return subtasks; } private void processing(String work) { String result = work.toUpperCase(); logger.info('This result - (' + result + ') - was processed by ' + Thread.currentThread().getName()); }}

上面的例子使用了二分法來打印字符串。

我們再看一個RecursiveTask<V>的例子:

public class CustomRecursiveTask extends RecursiveTask<Integer> { private int[] arr; private static final int THRESHOLD = 20; public CustomRecursiveTask(int[] arr) { this.arr = arr; } @Override protected Integer compute() { if (arr.length > THRESHOLD) { return ForkJoinTask.invokeAll(createSubtasks()) .stream() .mapToInt(ForkJoinTask::join) .sum(); } else { return processing(arr); } } private Collection<CustomRecursiveTask> createSubtasks() { List<CustomRecursiveTask> dividedTasks = new ArrayList<>(); dividedTasks.add(new CustomRecursiveTask( Arrays.copyOfRange(arr, 0, arr.length / 2))); dividedTasks.add(new CustomRecursiveTask( Arrays.copyOfRange(arr, arr.length / 2, arr.length))); return dividedTasks; } private Integer processing(int[] arr) { return Arrays.stream(arr) .filter(a -> a > 10 && a < 27) .map(a -> a * 10) .sum(); }}

和上面的例子很像,不過這里我們需要有返回值。

在ForkJoinPool中提交Task

有了上面的兩個任務(wù),我們就可以在ForkJoinPool中提交了:

int[] intArray= {12,12,13,14,15}; CustomRecursiveTask customRecursiveTask= new CustomRecursiveTask(intArray); int result = forkJoinPool.invoke(customRecursiveTask); System.out.println(result);

上面的例子中,我們使用invoke來提交,invoke將會等待任務(wù)的執(zhí)行結(jié)果。

如果不使用invoke,我們也可以將其替換成fork()和join():

customRecursiveTask.fork(); int result2= customRecursiveTask.join(); System.out.println(result2);

fork() 是將任務(wù)提交給pool,但是并不觸發(fā)執(zhí)行, join()將會真正的執(zhí)行并且得到返回結(jié)果。

本文的例子可以參考https://github.com/ddean2009/learn-java-concurrency/tree/master/forkjoin

到此這篇關(guān)于java中的fork join框架的使用的文章就介紹到這了,更多相關(guān)java fork join框架內(nèi)容請搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)!

標(biāo)簽: Java
相關(guān)文章:
主站蜘蛛池模板: 亚洲视频二区 | 欧美成人高清 | 亚洲一区二区在线 | 免费久久精品视频 | 国产精品美女久久久 | 亚洲三级在线 | 人操人人 | 黄色片免费在线观看 | 免费看爱爱视频 | 亚洲国产成人精品久久 | 午夜激情在线视频 | 成人在线精品视频 | 中文字幕成人av | 亚洲永久字幕 | 一区二区三区视频 | 亚洲高清av在线 | 亚洲国产日本 | 国产高潮好爽受不了了夜夜做 | 免费一级欧美在线观看视频 | 91久久久精品国产一区二区蜜臀 | 国产精品嫩草影院精东 | 毛片黄| 久久免费精品 | 亚洲精品在线看 | 精品国产一级 | 欧美精品久久久久久 | 国产精品日韩在线观看一区二区 | 久久久久国产精品午夜一区 | 九九精品在线 | 久久精品国产久精国产 | 夜夜骑首页 | 久日精品 | 久草视频网站 | 久久国产精品99久久久大便 | 免费一区在线 | av香港经典三级级 在线 | 成人在线观看免费 | 一区二区三区四区在线 | 久久一二区 | 四虎影院在线播放 | 嫩草伊人|