streamalism.org

1. Prehľad

Rámec fork / join bol predstavený v prostredí Java 7. Poskytuje nástroje, ktoré pomáhajú urýchliť paralelné spracovanie pokusom o použitie všetkých dostupných jadier procesora - čo sa dosahuje prostredníctvom prístupu rozdeliť a zvíťaziť.

V praxi to znamená rámec prvé „vidličky“, rekurzívne rozdelenie úlohy na menšie nezávislé čiastkové úlohy, kým nie sú také jednoduché, aby ich bolo možné vykonať asynchrónne.

Potom, začína časť „pripojiť sa“, v ktorom sú výsledky všetkých čiastkových úloh rekurzívne spojené do jedného výsledku, alebo v prípade úlohy, ktorá vracia neplatnosť, program jednoducho počká, kým sa vykoná každá čiastková úloha.

Na zabezpečenie efektívneho paralelného vykonávania používa platforma fork / join skupinu vlákien nazývaných ForkJoinPool, ktorá spravuje pracovné vlákna typu ForkJoinWorkerThread.

2. ForkJoinPool

The ForkJoinPool je srdcom rámca. Jedná sa o implementáciu ExecutorService ktorý spravuje pracovné vlákna a poskytuje nám nástroje na získanie informácií o stave a výkone fondu vlákien.

Pracovné vlákna môžu vykonávať naraz iba jednu úlohu, ale ForkJoinPool nevytvára samostatné vlákno pre každú jednotlivú podúlohu. Namiesto toho má každé vlákno v bazéne svoj vlastný obojstranný front (alebo deque, vyslovuje sa) paluba), v ktorom sú uložené úlohy.

Táto architektúra je nevyhnutná na vyrovnanie záťaže vlákna pomocou algoritmus kradnutia práce.

2.1. Algoritmus kradnutia práce

Zjednodušene - bezplatné vlákna sa snažia „ukradnúť“ prácu z problémov rušných vlákien.

Pracovné vlákno predvolene získava úlohy z hlavy svojho vlastného deque. Keď je prázdne, vlákno vezme úlohu z konca deque iného zaneprázdneného vlákna alebo z globálneho vstupného frontu, pretože práve tu sa pravdepodobne budú nachádzať najväčšie kúsky práce.

Tento prístup minimalizuje možnosť, že vlákna budú súťažiť o úlohy. Znižuje sa tiež počet opakovaní, kedy bude vlákno musieť hľadať prácu, pretože pracuje najskôr s najväčšími dostupnými časťami práce.

2.2. ForkJoinPool Instancia

V prostredí Java 8 je najpohodlnejším spôsobom, ako získať prístup k inštancii súboru ForkJoinPool je použitie jeho statickej metódy commonPool (). Ako jeho názov napovedá, poskytne sa tým odkaz na spoločný fond, ktorý je predvoleným fondom vlákien pre každého ForkJoinTask.

Podľa dokumentácie spoločnosti Oracle použitie preddefinovaného spoločného fondu znižuje spotrebu zdrojov, pretože to odrádza od vytvorenia samostatného fondu vlákien pre každú úlohu.

ForkJoinPool commonPool = ForkJoinPool.commonPool ();

Rovnaké správanie je možné dosiahnuť v prostredí Java 7 vytvorením a ForkJoinPool a priradiť ho k a verejná statická pole úžitkovej triedy:

public static ForkJoinPool forkJoinPool = nový ForkJoinPool (2);

Teraz je k nej ľahký prístup:

ForkJoinPool forkJoinPool = PoolUtil.forkJoinPool;

S ForkJoinPool’s konštruktory, je možné vytvoriť vlastný fond vlákien so špecifickou úrovňou paralelizmu, továrne na vlákna a obsluhy výnimiek. Vo vyššie uvedenom príklade má fond úroveň paralelizmu 2. To znamená, že fond bude používať 2 procesorové jadrá.

3. ForkJoinTask

ForkJoinTask je základný typ pre úlohy vykonávané vo vnútri ForkJoinPool. V praxi by sa mala rozšíriť jedna z jej dvoch podtried: RekurzívnaAkcia pre neplatný úlohy a RecursiveTask pre úlohy, ktoré vracajú hodnotu.Obaja majú abstraktnú metódu vypočítať () v ktorom je definovaná logika úlohy.

3.1. RecursiveAction - príklad

V nasledujúcom príklade predstavuje jednotku práce, ktorá sa má spracovať, a String zavolal pracovná záťaž. Pre demonštračné účely je to úloha nezmyselná: jednoducho zmení svoj vstup na veľký a zaznamená ho.

Na demonštráciu vidlicového správania rámu, príklad rozdelí úlohu, ak pracovná záťaž.length () je väčšie ako zadaný limitpomocou createSubtask () metóda.

Reťazec je rekurzívne rozdelený na podreťazce a vytvára sa CustomRecursiveTask inštancie, ktoré sú založené na týchto podreťazcoch.

Výsledkom je, že metóda vráti a Zoznam.

Zoznam sa predkladá ForkJoinPool pomocou invokeAll () metóda:

verejná trieda CustomRecursiveAction rozširuje RecursiveAction {súkromné ​​pracovné zaťaženie reťazca = ""; private static final int THRESHOLD = 4; private static Logger logger = Logger.getAnonymousLogger (); public CustomRecursiveAction (Pracovné zaťaženie reťazca) {this.workload = pracovné zaťaženie; } @Override protected void compute () {if (workload.length ()> THRESHOLD) {ForkJoinTask.invokeAll (createSubtasks ()); } else {spracovanie (pracovná záťaž); }} private List createSubtasks () {List subtasks = new ArrayList (); Reťazec partOne = workload.substring (0, workload.length () / 2); Reťazec partTwo = workload.substring (workload.length () / 2, workload.length ()); subtasks.add (new CustomRecursiveAction (partOne)); subtasks.add (new CustomRecursiveAction (partTwo)); vrátiť čiastkové úlohy; } spracovanie súkromných prázdnin (práca s reťazcami) {výsledok reťazca = work.toUpperCase (); logger.info ("Tento výsledok - (" + výsledok + ") - spracoval program" + Thread.currentThread (). getName ()); }}

Tento vzor sa dá použiť na vývoj vášho vlastného RekurzívnaAkcia triedy. Za týmto účelom vytvorte objekt, ktorý predstavuje celkové množstvo práce, zvolil vhodný prah, definoval metódu rozdelenia práce a definoval metódu vykonania práce.

3.2. RecursiveTask

Pre úlohy, ktoré vracajú hodnotu, je tu logika podobná, až na to, že výsledok pre každú podúlohu je zjednotený v jednom výsledku:

verejná trieda CustomRecursiveTask rozširuje RecursiveTask {private int [] arr; private static final int THRESHOLD = 20; public CustomRecursiveTask (int [] arr) {this.arr = arr; } @Override chránený Integer compute () {if (arr.length> THRESHOLD) {return ForkJoinTask.invokeAll (createSubtasks ()) .stream () .mapToInt (ForkJoinTask :: join) .sum (); } else {spracovanie vratky (arr); }} súkromná zbierka createSubtasks () {List splitTasks = new ArrayList (); splitTasks.add (nový CustomRecursiveTask (Arrays.copyOfRange (arr, 0, arr.length / 2)))); splitTasks.add (nový CustomRecursiveTask (Arrays.copyOfRange (arr, arr.length / 2, arr.length))); návrat rozdelenéÚlohy; } spracovanie súkromných celých čísel (int [] arr) {návrat Arrays.stream (arr) .filter (a -> a> 10 && a a * 10) .sum (); }}

V tomto príklade je práca predstavovaná poľom uloženým v prírastok pole CustomRecursiveTask trieda. The createSubtasks () metóda rekurzívne rozdeľuje úlohu na menšie časti práce, kým každá časť nie je menšia ako prahová hodnota. Potom invokeAll () metóda odošle čiastkové úlohy do spoločného fondu a vráti zoznam Budúcnosť.

Na spustenie exekúcie pripojiť sa () metóda sa volá pre každú podúlohu.

V tomto príklade sa to dosahuje pomocou Java 8 Stream API; the suma () metóda sa používa ako reprezentácia kombinácie čiastkových výsledkov do konečného výsledku.

4. Zadávanie úloh do ForkJoinPool

Na odosielanie úloh do oblasti vlákien je možné použiť niekoľko prístupov.

The Predložiť() alebo vykonať ()metóda (ich prípady použitia sú rovnaké):

forkJoinPool.execute (customRecursiveTask); int result = customRecursiveTask.join ();

The vzývať()metóda rozdeľuje úlohu a čaká na výsledok a nepotrebuje žiadne ručné spájanie:

int výsledok = forkJoinPool.invoke (customRecursiveTask);

The invokeAll () metóda je najpohodlnejším spôsobom zadania sekvencie ForkJoinTasks do ForkJoinPool. Berie úlohy ako parametre (dve úlohy, var args alebo kolekcia), forks potom vráti kolekciu Budúcnosť predmety v poradí, v akom boli vyrobené.

Prípadne môžete použiť samostatný vidlička() a pripojiť sa () metódy. The vidlička() metóda zadá úlohu do spoločného súboru, ale nespustí jej vykonávanie. The pripojiť sa () Na tento účel sa musí použiť metóda. V prípade RekurzívnaAkcia, pripojiť sa () nevracia nič iné ako nulový; pre RecursiveTask, vráti výsledok vykonania úlohy:

customRecursiveTaskFirst.fork (); result = customRecursiveTaskLast.join ();

V našom RecursiveTask príklad sme použili invokeAll () metóda na odoslanie postupnosti čiastkových úloh do spoločného fondu. Rovnakú prácu je možné vykonať vidlička() a pripojiť sa (), hoci to má dôsledky na usporiadanie výsledkov.

Aby nedošlo k nedorozumeniu, je zvyčajne dobré ho použiť invokeAll () - metóda na predloženie viac ako jednej úlohy ForkJoinPool.

5. Závery

Používanie rozhrania fork / join môže urýchliť spracovanie veľkých úloh, ale na dosiahnutie tohto výsledku by sa mali dodržiavať niektoré pokyny:

• Používajte čo najmenej združených vlákien - vo väčšine prípadov je najlepším rozhodnutím použiť jeden fond vlákien na aplikáciu alebo systém
• Použiť predvolený spoločný fond vlákien, ak nie je potrebné žiadne konkrétne ladenie
• Použite primeranú hranicu na štiepanie ForkJoinTask do čiastkových úloh
• Zabráňte blokovaniu vo vašomForkJoinTasks

Príklady použité v tomto článku sú k dispozícii v prepojenom úložisku GitHub.