streamalism.org

1. Pregled

Okvir fork / join predstavljen je u Javi 7. Pruža alate koji pomažu ubrzati paralelnu obradu pokušavajući koristiti sve dostupne procesorske jezgre - što je i postignuto kroz pristup podijeli i osvoji.

U praksi to znači da prvi okvir "račva", rekurzivno raščlanjivanje zadatka na manje neovisne podzadaće sve dok ne postanu dovoljno jednostavni za asinkrono izvršavanje.

Nakon toga, započinje dio "pridruživanja", u kojem se rezultati svih podzadataka rekurzivno spajaju u jedan rezultat, ili u slučaju zadatka koji se poništava, program jednostavno čeka dok se ne izvrši svaki podzadatak.

Da bi se osiguralo učinkovito paralelno izvršavanje, okvir fork / join koristi skup niti koji se naziva ForkJoinPool, koji upravlja radničkim nitima tipa ForkJoinWorkerThread.

2. ForkJoinPool

The ForkJoinPool je srce okvira. To je provedba ExecutorService koji upravlja radničkim nitima i pruža nam alate za dobivanje informacija o stanju i izvedbi spremišta niti.

Radničke niti mogu istovremeno izvršavati samo jedan zadatak, ali ForkJoinPool ne stvara zasebnu nit za svaki pojedini podzadatak. Umjesto toga, svaka nit u spremištu ima svoj dvostruki red (ili deque, izražen paluba) koja pohranjuje zadatke.

Ova je arhitektura vitalna za uravnoteženje radnog opterećenja niti uz pomoć algoritam krađe rada.

2.1. Algoritam krađe posla

Jednostavno rečeno, besplatne niti pokušavaju "ukrasti" posao od dequeova zauzetih niti.

Prema zadanim postavkama radnička nit prima zadatke iz glave vlastitog deque-a. Kada je prazan, nit preuzima zadatak iz repa deque-a druge zauzete niti ili iz globalnog reda unosa, jer će se tu vjerojatno nalaziti najveći dijelovi posla.

Ovaj pristup minimalizira mogućnost da se niti nadmeću za zadatke. Također smanjuje broj puta kada će nit morati potražiti posao, jer prvo radi na najvećim dostupnim dijelovima posla.

2.2. ForkJoinPool Instanciranje

U Javi 8, najprikladniji način za pristup instanci ForkJoinPool je koristiti njegovu statičku metodu commonPool (). Kao što mu samo ime govori, ovo će pružiti referencu na zajednički bazen, koji je zadani bazen niti za svaki ForkJoinTask.

Prema Oracleovoj dokumentaciji, korištenje unaprijed definiranog zajedničkog spremišta smanjuje potrošnju resursa, jer to obeshrabruje stvaranje zasebnog spremišta niti po zadatku.

ForkJoinPool commonPool = ForkJoinPool.commonPool ();

Isto se ponašanje može postići u Javi 7 stvaranjem a ForkJoinPool i dodijelivši ga a javna statika polje klase korisnosti:

javni statični ForkJoinPool forkJoinPool = novi ForkJoinPool (2);

Sada mu se može lako pristupiti:

ForkJoinPool forkJoinPool = PoolUtil.forkJoinPool;

S ForkJoinPool’s konstruktora, moguće je stvoriti prilagođeni bazen niti s određenom razinom paralelizma, tvornicom niti i rukovateljem iznimkama. U gornjem primjeru spremište ima razinu paralelizma 2. To znači da će spremište koristiti 2 procesorske jezgre.

3. ForkJoinTask

ForkJoinTask je osnovni tip za zadatke izvršene unutar ForkJoinPool. U praksi bi se trebala proširiti jedna od dvije podklase: Rekurzivno djelovanje za poništiti zadaci i Rekurzivna zadaća za zadatke koji vraćaju vrijednost.Oboje imaju apstraktnu metodu izračunati () u kojem je definirana logika zadatka.

3.1. RecursiveAction - primjer

U donjem primjeru jedinica rada koja se obrađuje predstavljena je s Niz pozvao opterećenje. U demonstracijske svrhe zadatak je besmislen: jednostavno unosi velika slova i bilježi ih.

Da bi demonstrirali forking ponašanje okvira, primjer dijeli zadatak ako opterećenje.length () je veći od navedenog pragakoristiti createSubtask () metoda.

String se rekurzivno dijeli na podnizove, stvarajući CustomRecursiveTask instance koje se temelje na tim podnizovima.

Kao rezultat, metoda vraća a Popis.

Popis se predaje na ForkJoinPool koristiti invokeAll () metoda:

javna klasa CustomRecursiveAction proširuje RecursiveAction {private String workload = ""; privatni statički konačni int PRAG = 4; privatni statički logger logger = Logger.getAnonymousLogger (); javno CustomRecursiveAction (radno opterećenje niza) {this.workload = radno opterećenje; } @Override protected void compute () {if (workload.length ()> THRESHOLD) {ForkJoinTask.invokeAll (createSubtasks ()); } else {obrada (radno opterećenje); }} privatni popis createSubtasks () {Podzadaci popisa = novi ArrayList (); String partOne = workload.substring (0, workload.length () / 2); String partTwo = workload.substring (workload.length () / 2, workload.length ()); subtasks.add (nova CustomRecursiveAction (partOne)); subtasks.add (nova CustomRecursiveAction (partTwo)); povrat podzadataka; } privatna obrada praznina (rad niza) {Rezultat niza = work.toUpperCase (); logger.info ("Ovaj rezultat - (" + rezultat + ") - obradio je" + Thread.currentThread (). getName ()); }}

Ovaj obrazac možete koristiti za razvoj vlastitog Rekurzivno djelovanje razreda. Da biste to učinili, stvorite objekt koji predstavlja ukupnu količinu posla, odaberite prikladni prag, definirajte metodu za podjelu djela i definirajte metodu za obavljanje posla.

3.2. Rekurzivna zadaća

Za zadatke koji vraćaju vrijednost, logika je ovdje slična, osim što je rezultat za svaki podzadatak objedinjen u jedan rezultat:

javna klasa CustomRecursiveTask proširuje RecursiveTask {private int [] arr; privatni statički konačni int PRAG = 20; javni CustomRecursiveTask (int [] arr) {this.arr = arr; } @Override protected Integer compute () {if (arr.length> THRESHOLD) {return ForkJoinTask.invokeAll (createSubtasks ()) .stream () .mapToInt (ForkJoinTask :: join) .sum (); } else {obrada povrata (arr); }} privatna kolekcija createSubtasks () {Popis podijeljenih zadataka = novi ArrayList (); sharedTasks.add (novi CustomRecursiveTask (Arrays.copyOfRange (arr, 0, arr.length / 2))); sharedTasks.add (novi CustomRecursiveTask (Arrays.copyOfRange (arr, arr.length / 2, arr.length))); povratak splitTasks; } privatna obrada cijelog broja (int [] arr) {return Arrays.stream (arr) .filter (a -> a> 10 && a a * 10) .sum (); }}

U ovom primjeru djelo je prikazano nizom pohranjenim u dolazak polje CustomRecursiveTask razred. The createSubtasks () metoda rekurzivno dijeli zadatak na manje dijelove rada sve dok svaki dio nije manji od praga. Onda invokeAll () metoda predaje podzadaće zajedničkom spremištu i vraća popis Budućnost.

Da bi pokrenuo izvršenje, pridružiti() metoda se poziva za svaki podzadatak.

U ovom primjeru to se postiže korištenjem Jave 8 Stream API; the iznos() metoda koristi se kao prikaz kombiniranja podrezultata u konačni rezultat.

4. Slanje zadataka na ForkJoinPool

Za predavanje zadataka u spremište niti može se koristiti nekoliko pristupa.

The podnijeti() ili izvršiti()metoda (slučajevi njihove upotrebe su isti):

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

The prizvati ()metoda forksira zadatak i čeka rezultat i ne treba ručno spajanje:

int rezultat = forkJoinPool.invoke (customRecursiveTask);

The invokeAll () metoda je najprikladniji način slanja niza od ForkJoinTasks prema ForkJoinPool. Zadatke uzima kao parametre (dva zadatka, var args ili zbirku), a zatim forks vraća kolekciju Budućnost predmeti redoslijedom kojim su proizvedeni.

Možete koristiti i zasebne vilica () i pridružiti() metode. The vilica () metoda predaje zadatak spremištu, ali ne pokreće njegovo izvršenje. The pridružiti() U tu svrhu mora se koristiti metoda. U slučaju Rekurzivno djelovanje, pridružiti() ne vraća ništa osim null; za Rekurzivna zadaća, vraća rezultat izvršenja zadatka:

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

U našem Rekurzivna zadaća primjer smo koristili invokeAll () metoda za podnošenje niza podzadataka spremištu. Može se raditi isti posao vilica () i pridružiti(), iako to ima posljedice na redoslijed rezultata.

Kako bi se izbjegla zabuna, općenito je dobro koristiti invokeAll () metoda za podnošenje više zadataka ForkJoinPool.

5. Zaključci

Upotreba okvira fork / join može ubrzati obradu velikih zadataka, ali da bi se postigao taj ishod treba slijediti neke smjernice:

• Upotrijebite što manje spremišta niti - u većini je slučajeva najbolja odluka koristiti jednu spremu niti po aplikaciji ili sustavu
• Koristite zadano spremište zajedničkih niti, ako nije potrebno posebno podešavanje
• Koristite razumni prag za cijepanje ForkJoinTask u podzadaće
• Izbjegavajte blokiranje na vašemForkJoinTasks

Primjeri korišteni u ovom članku dostupni su u povezanom spremištu GitHub.