Duboko razumevanje Java konkurentnog programiranja — Fork/Join okvir
Zadaci u polju konkurentnog programiranja mogu se podeliti u tri vrste: jednostavni paralelni zadaci, agregacioni zadaci i grupni paralelni zadaci, kao na slici.
Pored ovih modela postoji još jedan model zadataka poznat kao „podeli pa vladaj“. Podeli pa vladaj je način razmišljanja i obrazac za rešavanje složenih problema: konkretno, složen problem se rastavlja na više sličnih potproblema, koji se zatim dalje rastavljaju na još manje potprobleme, sve dok svaki potproblem ne postane dovoljno jednostavan da se neposredno reši.
Teorijski, svaki problem odgovara jednom zadatku, pa je „podeli pa vladaj“ zapravo podela i organizacija zadataka. Obrazac „podeli pa vladaj“ ima široku primenu u mnogim oblastima. Na primer, u algoritmima često koristimo algoritme podeli pa vladaj (merge sort i quick sort spadaju u tu kategoriju, a i binarna pretraga je algoritam podeli pa vladaj). U oblasti velikih podataka, i ideja koja stoji iza MapReduce okvira za izračunavanje zasnovana je na podeli pa vladaj.
Zbog ovakve opšte primenljivosti modela „podeli pa vladaj“, Java paket za konkurentnost pruža paralelni okvir za izračunavanje pod imenom Fork/Join, namenjen podršci primenama tog modela zadataka.
Šta je model zadataka „podeli pa vladaj“
Model zadataka „podeli pa vladaj“ ima dve faze: jedna je rastavljanje zadataka — iterativno rastavljanje zadatka na podzadatke dok podzadatak ne može neposredno da izračuna rezultat; druga je spajanje rezultata — postepeno spajanje rezultata podzadataka dok se ne dobije konačni rezultat. Ispod je pojednostavljen prikaz modela zadataka „podeli pa vladaj“ kojeg možete pratiti.
U ovom modelu zadatak i podzadaci koji nastaju rastavljanjem imaju sličnost, koja se obično ogleda u tome što je algoritam zadatka i podzadatka isti, ali se razlikuje obim podataka za izračunavanje. Za probleme sa ovom sličnošću najčešće koristimo rekurzivne algoritme.
Upotreba Fork/Join
Fork/Join je okvir za paralelno izračunavanje, prvenstveno namenjen podršci modelu zadataka „podeli pa vladaj“. U ovom okviru Fork predstavlja rastavljanje zadatka, a Join spajanje rezultata.
Fork/Join okvir se uglavnom sastoji od dva dela: bazena niti za zadatke „podeli pa vladaj“ ForkJoinPool i zadatka „podeli pa vladaj“ ForkJoinTask.
Odnos između ova dva dela je sličan odnosu između ThreadPoolExecutor i Runnable — oba služe za podnošenje zadataka bazenu niti, samo što zadatak „podeli pa vladaj“ ima svoj specifičan tip ForkJoinTask.
ForkJoinTask je apstraktna klasa sa mnogo metoda; najsrbinskiji su fork() i join(). Metod fork služi za asinhrono izvršavanje podzadatka, dok metod join blokira trenutnu nit i čeka rezultat podzadatka.
ForkJoinTask ima dve potklase: RecursiveAction i RecursiveTask.
Po imenu se vidi da obe obrađuju zadatke „podeli pa vladaj“ rekurzivno. Obe potklase definišu apstraktan metod compute(); razlika je u tome što compute kod RecursiveAction nema povratnu vrednost, dok compute kod RecursiveTask ima povratnu vrednost. Obe potklase su apstraktne, pa pri upotrebi treba kreirati sopstvene potklase koje proširuju funkcionalnost.
Naredno, iskoristićemo Fork/Join paralelni okvir za izračunavanje Fibonačijevog niza (primer koda potiče iz zvaničnog Java primera).
Najpre treba kreirati bazen niti ForkJoinPool i zadatak „podeli pa vladaj“ Fibonacci za izračunavanje Fibonačijevog niza. Zatim, pozivom metoda invoke() bazena ForkJoinPool pokreće se zadatak.
Pošto izračunavanje Fibonačijevog niza zahteva povratnu vrednost, naša klasa Fibonacci nasleđuje RecursiveTask. Zadatak Fibonacci mora implementirati metod compute; logika u tom metodu je vrlo slična običnom načinu izračunavanja Fibonačijevog niza, osim što se pri izračunavanju Fibonacci(n - 1) koristi asinhroni podzadatak, što se postiže naredbom f1.fork().
@Slf4j
public class ForkJoinDemo {
// 1. Ulazna tačka izvršavanja
public static void main(String[] args) {
int n = 20;
// Da bi se pratilo ime podnite, potrebno je prepisati metod ForkJoinWorkerThreadFactory
final ForkJoinPool.ForkJoinWorkerThreadFactory factory = pool -> {
final ForkJoinWorkerThread worker = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
worker.setName("my-thread" + worker.getPoolIndex());
return worker;
};
//Kreira bazen zadataka podeli pa vladaj, može se pratiti ime niti
ForkJoinPool forkJoinPool = new ForkJoinPool(4, factory, null, false);
// Brz način kreiranja ForkJoinPool-a
// ForkJoinPool forkJoinPool = new ForkJoinPool(4);
//Kreira zadatak podeli pa vladaj
Fibonacci fibonacci = new Fibonacci(n);
//Poziva invoke za pokretanje zadatka podeli pa vladaj
Integer result = forkJoinPool.invoke(fibonacci);
log.info("Rezultat Fibonačijevog broja {} je {}", n, result);
}
}
// 2. Definiše zadatak rastavljanja, napiše logiku rastavljanja
@Slf4j
class Fibonacci extends RecursiveTask<Integer> {
final int n;
Fibonacci(int n) {
this.n = n;
}
@Override
public Integer compute() {
//Slično rekurziji, definiše najmanju jedinicu koja se može izračunati
if (n <= 1) {
return n;
}
// Ko želi da vidi ispis imena podnite može otkomentarisati donju liniju
//log.info(Thread.currentThread().getName());
Fibonacci f1 = new Fibonacci(n - 1);
// Rastavlja na podzadatak
f1.fork();
Fibonacci f2 = new Fibonacci(n - 2);
// f1.join čeka rezultat podzadatka
return f2.compute() + f1.join();
}
}Pokretanjem programa dobićemo sledeći rezultat:
17:29:10.336 [main] INFO tech.shuyi.javacodechip.forkjoinpool.ForkJoinDemo - Rezultat Fibonačijevog broja 20 je 6765ForkJoinPool
Ključna komponenta Fork/Join paralelnog izračunavanja je ForkJoinPool. Ukratko ćemo predstaviti princip rada ForkJoinPool-a.
Kada preko metoda invoke ili submit ForkJoinPool-a podnesemo zadatak, ForkJoinPool prema određenom pravilu rutiranja dodeljuje zadatak u red zadataka. Ako se tokom izvršavanja zadatka kreiraju podzadaci, podzadaci se podnose u red zadataka odgovarajuće radne niti.
ForkJoinPool ima član u obliku niza workQueue[], koji odgovara nizu redova, pri čemu svaki red odgovara jednoj niti potrošača. Zadatak ubačen u bazen niti prosleđuje se prema specifičnom pravilu.
Kada je red zadataka radne niti prazan, da li ona nema šta da radi?
Nema. ForkJoinPool uvodi mehanizam poznat kao „krađa zadataka“ (work stealing). Kada je radna nit slobodna, može „ukrasti“ zadatak iz reda zadataka druge radne niti.
Na primer, na slici ispod je red zadataka niti T2 već prazan, pa ona može ukrasti zadatak iz reda niti T1. Tako sve radne niti mogu ostati zauzete.
Redovi zadataka u ForkJoinPool-u su realizovani kao obostrani redovi. Radna nit uzima zadatke s jednog kraja reda, dok „krađa zadataka“ vrši potrošnju s drugog kraja. Ovakav dizajn izbegava mnoge nepotrebne sukobe oko podataka.
Poređenje sa ThreadPoolExecutor
ForkJoinPool i ThreadPoolExecutor imaju mnogo sličnosti — oba su bazeni niti i oba služe za izvršavanje zadataka. Ali između njih postoje i mnoge razlike.
Pre svega, ForkJoinPool koristi mehanizam „krađa posla“, dok ThreadPoolExecutor koristi mehanizam „ponovnog korišćenja rada“. Oba mehanizma imaju svoje prednosti i mane: ForkJoinPool je jak u iskorišćavanju višejezgrenih sposobnosti CPU-a, dok je ThreadPoolExecutor jak u izbegavanju prebacivanja konteksta među nitima.
Drugo, ForkJoinPool koristi model zadataka „podeli pa vladaj“, dok ThreadPoolExecutor koristi model jednostavnih paralelnih zadataka. ForkJoinPool je pogodan za zadatke „podeli pa vladaj“, a ThreadPoolExecutor za jednostavne paralelne zadatke.
Konačno, ForkJoinPool koristi LIFO red zadataka, dok ThreadPoolExecutor koristi FIFO red zadataka. ForkJoinPool izbegava sukobe oko podataka, dok ThreadPoolExecutor osigurava redosled zadataka.
Pretpostavimo: želimo da izračunamo zbir od 1 do 100 miliona; da bismo ubrzali izračunavanje, naravno pomišljamo na princip „podeli pa vladaj“ iz algoritama — podelimo 100 miliona brojeva na 10 000 zadataka, gde svaki zadatak izračunava zbir 10 000 brojeva, čime se iskorišćava konkurentna izračunavajuća moć CPU-a i skraćuje vreme izračunavanja.
Pošto ThreadPoolExecutor putem Future može dobiti rezultat izvršavanja, korišćenje ThreadPoolExecutor-a je takođe izvodljivo.
Naravno, ForkJoinPool takođe može da se iskoristi. U nastavku ćemo realizovati oba načina i uporediti ih.
Bez obzira na način realizacije, opšta ideja je:
- Prema broju niti N u bazenu, podeliti 100 miliona brojeva na N jednakih delova, a zatim ih predati bazenu na izračunavanje.
- Svaki zadatak izračunavanja putem interfejsa Future dobavlja rezultat, koji se zatim sumira.
Najpre ćemo realizovati preko ThreadPoolExecutor-a.
Prvo definišemo interfejs Calculator koji predstavlja radnju izračunavanja zbira brojeva:
public interface Calculator {
/**
* Sumira sve prosleđene brojeve
*
* @param numbers
* @return zbir
*/
long sumUp(long[] numbers);
}Zatim definišemo klasu koja koristi ThreadPoolExecutor bazen niti:
public class ExecutorServiceCalculator implements Calculator {
private int parallism;
private ExecutorService pool;
public ExecutorServiceCalculator() {
// Broj jezgara CPU-a; podrazumevano se koristi broj jezgara CPU-a
parallism = Runtime.getRuntime().availableProcessors();
pool = Executors.newFixedThreadPool(parallism);
}
// 1. Nit koja obrađuje zadatak izračunavanja
private static class SumTask implements Callable<Long> {
private long[] numbers;
private int from;
private int to;
public SumTask(long[] numbers, int from, int to) {
this.numbers = numbers;
this.from = from;
this.to = to;
}
@Override
public Long call() {
long total = 0;
for (int i = from; i <= to; i++) {
total += numbers[i];
}
return total;
}
}
// 2. Realizacija jezgra poslovne logike
@Override
public long sumUp(long[] numbers) {
List<Future<Long>> results = new ArrayList<>();
// 2.1 Podela brojeva
// Zadatak se deli na n delova i predaje n niti; 4 jezgra znači podela na 4 jednaka dela
// a zatim se svaki deo predaje jednoj SumTask niti na obradu
int part = numbers.length / parallism;
for (int i = 0; i < parallism; i++) {
int from = i * part; //početna pozicija
int to = (i == parallism - 1) ? numbers.length - 1 : (i + 1) * part - 1; //završna pozicija
//predaje bazenu niti na izračunavanje
results.add(pool.submit(new SumTask(numbers, from, to)));
}
// 2.2 Blokira i čeka rezultate
// Rezultati svake niti se sabiraju da bi se dobio konačni rezultat; get() je blokirajući
// Moguće poboljšanje: upotrebom CompletableFuture, nove osobine JDK-a 1.8
long total = 0L;
for (Future<Long> f : results) {
try {
total += f.get();
} catch (Exception ignore) {
}
}
return total;
}
}Kao što je prikazano, realizovali smo klasu SumTask za izračunavanje jednog zadatka, u kojoj se vrši akumulacija brojeva. Zatim, u metodu sumUp(), delim 100 miliona brojeva, predajem ih bazenu niti na izračunavanje i konačno blokiram i čekam rezultate, koje akumuliram.
Pokretanjem gornjeg koda glatko dobijamo konačni rezultat:
Trajanje: 10ms
Rezultat: 50000005000000Zatim ćemo iskoristiti ForkJoinPool.
Najpre realizujemo SumTask koji nasleđuje apstraktnu klasu RecursiveTask i u metodu compute definišemo logiku rastavljanja i izračunavanja. Konačno, u metodu sumUp pozivamo bazen za izračunavanje:
public class ForkJoinCalculator implements Calculator {
private ForkJoinPool pool;
// 1. Definiše logiku izračunavanja
private static class SumTask extends RecursiveTask<Long> {
private long[] numbers;
private int from;
private int to;
public SumTask(long[] numbers, int from, int to) {
this.numbers = numbers;
this.from = from;
this.to = to;
}
//Ovo je ključni metod ForkJoin-a: vrši rastavljanje zadatka; kvalitet rastavljanja određuje efikasnost
@Override
protected Long compute() {
// Kada je broj brojeva za izračunavanje manji od 6, koristi se for petlja
if (to - from < 6) {
long total = 0;
for (int i = from; i <= to; i++) {
total += numbers[i];
}
return total;
} else {
// Inače, zadatak se deli na dva dela i rekurzivno rastavlja (koliko delova zavisi od situacije)
int middle = (from + to) / 2;
SumTask taskLeft = new SumTask(numbers, from, middle);
SumTask taskRight = new SumTask(numbers, middle + 1, to);
taskLeft.fork();
taskRight.fork();
return taskLeft.join() + taskRight.join();
}
}
}
public ForkJoinCalculator() {
// Može se koristiti i zajednički bazen ForkJoinPool.commonPool():
// pool = ForkJoinPool.commonPool()
pool = new ForkJoinPool();
}
@Override
public long sumUp(long[] numbers) {
Long result = pool.invoke(new SumTask(numbers, 0, numbers.length - 1));
pool.shutdown();
return result;
}
}Pokretanjem gornjeg koda dobijamo:
Trajanje: 860ms
Rezultat: 50000005000000Poređenjem realizacija ThreadPoolExecutor i ForkJoinPool vidimo da obje imaju logiku podele zadataka i logiku konačnog spajanja brojeva. Ali ForkJoinPool u odnosu na ThreadPoolExecutor vrši neka pakovanja na nivou realizacije, na primer:
- Ne mora ručno da se dobavljaju rezultati podzadataka, već se koristi metod join koji neposredno dobavlja rezultat.
- Logika podele zadataka je obuhvaćena u implementacionoj klasi RecursiveTask, umesto da bude izložena spolja.
Zato za paralelne zadatke izračunavanja bez zavisnosti između roditeljskih i podzadataka, a kod kojih se žele dobiti rezultati podzadataka, može se iskoristiti ForkJoinPool. U takvoj situaciji, ForkJoinPool je pre svega pogodniji zbog jednostavnije realizacije koda i boljeg pakovanja.
Simulacija MapReduce za brojanje reči
MapReduce je model programiranja, a ujedno i okvir za obradu i generisanje velikih skupova podataka. Potiče od Google-a, a služi za podršku distribuiranom izračunavanju nad velikim skupovima podataka. Okvir se uglavnom sastoji od dva koraka: korak Map i korak Reduce — po čemu i nosi ime.
Fork/Join paralelni okvir za izračunavanje obično se koristi za realizaciju uvodnog programa u učenju MapReduce-a, koji broji koliko se puta svaka reč pojavljuje u datoteci.
Najpre možemo rekurzivno, metodom polovljenja, rastaviti datoteku na manje delove sve dok svaki deo ne sadrži samo jednu liniju podataka. Zatim se u svakom delu prebroje reči i rezultati se postepeno agregiraju. Možete se poslužiti ranije pomenutim pojednostavljenim dijagramom modela „podeli pa vladaj“ da biste razumeli taj proces.
Sada krenimo sa realizacijom. Sledeći kod koristi niz stringova String[] fc za simulaciju sadržaja datoteke, gde svaki element odgovara jednoj liniji u datoteci. Ključni kod nalazi se u metodu compute(), koji je rekurzivni metod. On polovinu podataka prosleđuje rekurzivnom zadatku obrade putem fork-a (ključni kod: mr1.fork()), dok drugu polovinu obrađuje rekurzivno u trenutnom zadatku (mr2.compute()).
import java.util.concurrent.RecursiveTask;
public class WordCountTask extends RecursiveTask<Integer> {
private final String[] fc;
private final int start, end;
public WordCountTask(String[] fc, int start, int end) {
this.fc = fc;
this.start = start;
this.end = end;
}
@Override
protected Integer compute() {
if (end - start <= 1) {
// Broji reči u jednoj liniji
return countWords(fc[start]);
} else {
int mid = (start + end) / 2;
WordCountTask mr1 = new WordCountTask(fc, start, mid);
mr1.fork();
WordCountTask mr2 = new WordCountTask(fc, mid, end);
int result2 = mr2.compute();
int result1 = mr1.join();
// Agregira rezultate
return result1 + result2;
}
}
private int countWords(String line) {
String[] words = line.split(" ");
return words.length;
}
}Ovaj primer je pojednostavljenje Fork/Join modela; u pravom MapReduce okviru postoji još koraka — podela podataka, faza mapiranja, faza redukcije i drugo. Ali kroz ovaj primer možete steći početni uvid u to kako iskoristiti Fork/Join okvir za obradu sličnih zadataka.
Kratak pregled
Fork/Join paralelni okvir za izračunavanje pre svega rešava zadatke „podeli pa vladaj“. Jezgro te ideje je „podeli pa vladaj“: veliki zadatak se rastavlja na male podzadatke, a zatim se rezultati podzadataka agregiraju u konačni rezultat. Ovaj proces je veoma sličan MapReduce-u u obradi velikih podataka, pa Fork/Join možete posmatrati kao jednostrojnu (lokalnu) verziju MapReduce-a.
Ključna komponenta Fork/Join okvira je ForkJoinPool. ForkJoinPool podržava mehanizam krađe zadataka, koji omogućava da opterećenje svih niti bude uglavnom izbalansirano, tako da ne dolazi do situacije da su neke niti prezauzete, a druge nerade; performanse su stoga odlične.
Java 1.8 pruža Stream API čiji paralelni tokovi takođe počivaju na ForkJoinPool-u. Treba napomenuti da podrazumevano svi paralelni tokovi dele jedan isti ForkJoinPool, čiji je podrazumevani broj niti jednak broju jezgara CPU-a; ako su svi paralelni tokovi CPU-intenzivni izračunavanja, to nije problem, ali ako postoje I/O-intenzivni paralelni tokovi, jedan spori I/O izračun može usporiti performanse celog sistema.
Zato se preporučuje korišćenje različitih ForkJoinPool bazena za različite vrste zadataka izračunavanja.
Urednik: Chenmo Wang Er, deo sadržaja potiče iz ovog članka: Ideja podeli pa vladaj — Forkjoin, a deo iz ovog članka prijatelja Chen Shuyi-ja, koji je vrlo sadržajan i toplo se preporučuje. Deo slika potiče iz članka prijatelja „Ri Gong Yi Bing“.
Na GitHub-u je konačno stigao drugi PDF „Mali priručnik o konkurentnom programiranju“open-source baze znanja sa preko 17000 zvezdica „Ergov put ka naprednom Javom“! Obuhvata osnovne pojmove i načine upotrebe niti, memorijski model Java-e, synchronized, volatile, CAS, AQS, ReentrantLock, bazene niti, konkurentne kontejnere, ThreadLocal, model proizvođač-potrošač i druge teme koje su obavezne za intervjue i razvoj — ukupno preko 150 000 reči i preko 200 ručno nacrtanih ilustracija, što se može opisati kao pristupačno i sa humorom... Više detalja: Odlično, Ergov put ka naprednom konkurentnom programiranju.pdf