24 ilustracije koje vas vode kroz potpuno razumevanje Java bazena niti
Dobro, konačno smo stigli do Java bazena niti — to je veoma važan deo Java konkurentnog programiranja. Danas ćemo kroz slike i tekst potpuno razumeti princip rada bazena niti i kako u realnim projektima napraviti bazen niti prilagođen sopstvenom biznisu.
Jedan: Šta je bazen niti
Bazen niti je zapravo jedna realizacija tehnologije udruživanja (pooling). Suština tehnologije udruživanja je u ponovnom korišćenju resursa, čime se izbegavaju performansni troškovi neprestanog kreiranja i uništavanja resursa. Bazen niti može upravljati skupom niti, tako da nit nakon izvršenja zadatka ne bude uništena, već nastavi da obrađuje druge zadatke koje su joj prosledile druge niti.
Prednosti korišćenja bazena niti:
- Smanjenje potrošnje resursa. Ponovnim korišćenjem već kreiranih niti se smanjuje potrošnja koja nastaje kreiranjem i uništavanjem niti.
- Povećanje brzine odziva. Kada zadatak stigne, ne mora čekati na kreiranje niti već može odmah početi sa izvršenjem.
- Povećanje upravljivosti niti. Nit je retki resurs; neograničeno kreiranje ne samo da troši sistemske resurse, već i smanjuje stabilnost sistema. Bazen niti omogućava jedinstvenu alokaciju, podešavanje i nadzor.
Dva: Konstrukcija bazena niti
Java uglavnom kreira bazen niti tako što izgradi ThreadPoolExecutor. Hajde da pogledamo kako se bazen niti konstruiše.
Konstruktor ThreadPoolExecutor-a:
- corePoolSize: broj jezgrenih niti u bazenu niti koje obavljaju posao.
- maximumPoolSize: maksimalni broj niti; najveći broj niti koje bazen niti sme da kreira.
- keepAliveTime: vreme preživljavanja niti kreiranih iznad corePoolSize ili maksimalno vreme preživljavanja svih niti — zavisi od konfiguracije.
- unit: vremenska jedinica za keepAliveTime.
- workQueue: red zadataka; to je blokirajući red — kada broj niti dostigne broj jezgrenih niti, zadaci se skladište u blokirajući red.
- threadFactory: fabrika koju bazen niti interno koristi za kreiranje niti.
- handler: strategija odbijanja; kada je red pun i broj niti dostigne maksimalan, ovaj metod se poziva da obradi zadatak.
Konstrukcija bazena niti je zapravo veoma jednostavna — prosleđuje se gomila parametara, a zatim se vrše jednostavne dodele.
Tri: Princip rada bazena niti
Pošto smo prošli kroz ključne konstruktorske parametre bazena niti, hajde da objasnimo kako ti parametri rade u okviru bazena niti.
Kako izgleda bazen niti odmah nakon kreiranja? Kao na slici ispod:
Tačno tako — u tek kreiranom bazenu niti postoji samo blokirajući red prosleđen konstruktorom, bez ijedne niti. Ako želite pre izvršenja da kreirate sve jezgrene niti, to možete postići metodom prestartAllCoreThreads; podrazumevano nema nijedne niti.
Šta se desi kada jedna nit, preko metode execute, prosledi zadatak?
Prvo se proverava da li je broj niti u trenutnom bazenu niti manji od broja jezgrenih niti, odnosno od parametra corePoolSize prosleđenog pri konstrukciji.
Ako jeste, onda se preko ThreadFactory direktno kreira nit da izvrši taj zadatak, kao na slici:
Kada se zadatak izvrši, nit ne izlazi, već ide u blokirajući red po zadatak, kao na slici ispod:
Zatim, ako se prosledi još jedan zadatak, njega čekaju iste gore opisane procedure — proverava se da li je broj manji od jezgrenog, i ako jeste, ponovo se kreira nit za taj zadatak; i nakon izvršenja nit ide po zadatak iz blokirajućeg reda.
Ovde postoji jedan detalj: prilikom prosleđivanja zadatka, čak i ako niti iz bazena ne mogu da dohvate zadatak iz blokirajućeg reda, a je broj niti u bazenu i dalje manji od jezgrenog, kreiranje nove niti se nastavlja, umesto ponovnog korišćenja postojeće.
Ako broj niti u bazenu više nije manji od jezgrenog? Tada se pokušava sa smeštanjem zadatka u blokirajući red; nakon uspešnog enqueue-ovanja, kao na slici:
Tako blokirane niti mogu dohvatiti zadatak.
Međutim, kako zadataka bude sve više, red se popuni i enqueue-ovanje ne uspe — šta onda?
Tada se proverava da li je broj niti u bazenu manji od maksimalnog broja niti, odnosno od parametra maximumPoolSize prosleđenog pri konstrukciji.
Ako jeste manji od maksimalnog, kreira se ne-jezgrena nit koja izvršava prosleđeni zadatak, kao na slici:
Prema tome, čak i ako u redu ima zadataka, novokreirana nit ipak prvo obrađuje taj prosleđeni zadatak, umesto da iz reda dohvati postojeće zadatke — iz ovoga se vidi da prvo prosleđeni zadatak ne mora nužno biti prvo izvršen.
Ako je broj niti dostigao maksimalni broj, šta onda?
Tada se izvršava strategija odbijanja, odnosno objekat RejectedExecutionHandler prosleđen pri konstrukciji bazena niti, koji obrađuje taj zadatak.
JDK dolazi sa 4 ugrađene implementacije RejectedExecutionHandler-a:
- AbortPolicy: odbacuje zadatak i baca izuzetak izvršenja
- CallerRunsPolicy: zadatak izvršava nit koja ga je prosledila
- DiscardPolicy: odbacuje zadatak, ali bez bacanja izuzetka
- DiscardOldestPolicy: izbacuje zadatak koji je prvi ušao u red, a zatim ponovo prosleđuje zadatak
Ako se pri kreiranju bazena niti ne navede strategija odbijanja, podrazumevana je AbortPolicy.
Naravno, možete sami implementirati interfejs RejectedExecutionHandler — na primer, zadatak možete sačuvati u bazu ili keš, čime ga kasnije možete preuzeti iz baze ili keša.
Ovde primećujemo da smo u gore opisanom procesu izvršenja obradili parametre corePoolSize, maximumPoolSize, workQueue, threadFactory i handler, a da nismo obradili keepAliveTime i unit (jedinica vremena za keepAliveTime), pa kako keepAliveTime zapravo deluje — to ostavljamo za kasnije.
Pošto smo prošli kroz ceo proces izvršenja, pogledajmo sada kako je implementiran kod metode execute.
public void execute(Runnable command) {
// prvo proverava da li je prosleđeni zadatak null; ako jeste, baca NullPointerException
if (command == null)
throw new NullPointerException();
// preuzima trenutno stanje bazena niti (ctl je AtomicInteger koji sadrži stanje bazena i broj radnih niti)
int c = ctl.get();
// 1. proverava da li je broj radnih niti manji od broja jezgrenih niti (corePoolSize)
if (workerCountOf(c) < corePoolSize) {
// ako je manji od jezgrenog, pokušava da doda novu radnu nit koja će izvršiti prosleđeni zadatak
// addWorker proverava stanje bazena i broj radnih niti i odlučuje da li zaista dodati novu nit
if (addWorker(command, true))
return;
// ponovo preuzima stanje bazena jer se u pokušaju dodavanja stanje moglo promeniti
c = ctl.get();
}
// 2. pokušava da doda zadatak u red zadataka
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// dvostruka provera stanja bazena
if (! isRunning(recheck) && remove(command)) // ako je bazen zaustavljen, uklanja zadatak iz reda
reject(command);
// ako bazen radi, ali je broj radnih niti 0, pokušava da doda novu radnu nit
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 3. ako je red zadataka pun, pokušava da doda novu ne-jezgrenu radnu nit za izvršenje zadatka
else if (!addWorker(command, false))
// ako se nova radna nit ne može dodati (možda zato što je bazen zaustavljen ili je dostignut maksimalni broj niti), zadatak se odbija
reject(command);
}workerCountOf(c)<corePoolSize: proverava da li je broj manji od jezgrenog; ako jeste, ide se u metod addWorker koji dodaje nit i izvršava zadatak.workQueue.offer(command): pokušava da doda zadatak u blokirajući red. Ako dodavanje ne uspe, ponovo se poziva addWorker u pokušaju dodavanja ne-jezgrene niti za izvršenje zadatka; ako ni to ne uspe, poziva sereject(command)kojim se taj zadatak odbija.
Još jedan dijagram koji sumira tok izvršenja metode execute:
Četiri: Kako bazen niti ostvaruje ponovno korišćenje niti
Ključna funkcija bazena niti je ponovno korišćenje niti — kako on to radi?
Nit se interno u bazenu niti umotava u Worker objekat.
Worker nasleđuje AQS, pa ima izvesna svojstva brave.
Metod kojim se kreira nit za izvršenje zadatka je gore pomenuti addWorker. Pri kreiranju Worker objekta, nit i zadatak se zajedno umotavaju u Worker, a zatim se poziva metod runWorker kojim nit izvršava zadatak; pogledajmo metod runWorker.
final void runWorker(Worker w) {
// preuzima trenutnu radnu nit
Thread wt = Thread.currentThread();
// iz Workera izdvaja prvi zadatak
Runnable task = w.firstTask;
w.firstTask = null;
// otključava Workera (dozvoljava prekid)
w.unlock();
boolean completedAbruptly = true;
try {
// dok god postoji zadatak za izvršenje ili se iz reda zadataka može dohvatiti zadatak, radna nit nastavlja sa radom
while (task != null || (task = getTask()) != null) {
// zaključava Workera, osiguravajući da tokom izvršenja zadatka druge niti ne ometaju
w.lock();
// ako se bazen zaustavlja, osigurava se da je nit prekinuta
// ako nit nije prekinuta, a bazen je dostigao stanje zaustavljanja, nit se prekida
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// pre izvršenja zadatka mogu se umetnuti proizvoljne operacije
beforeExecute(wt, task);
Throwable thrown = null;
try {
// stvarno izvršenje zadatka
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
// nakon izvršenja zadatka mogu se umetnuti proizvoljne operacije
afterExecute(task, thrown);
}
} finally {
// prazni zadatak i ažurira brojač završenih zadataka
task = null;
w.completedTasks++;
// otključava Workera
w.unlock();
}
}
completedAbruptly = false;
} finally {
// nakradnja obrada izlaska radne niti
processWorkerExit(w, completedAbruptly);
}
}Ovde se može naći i razlog zašto nit nakon izvršenja zadatka ne izlazi — runWorker interno koristi beskonačnu while petlju; kada se prvi zadatak izvrši, stalno se preko metode getTask pokušava dohvatiti sledeći zadatak; sve dok se zadatak može dohvatiti, poziva se metod run koji nastavlja sa izvršenjem — to je glavni razlog zašto se nit može ponovo koristiti.
Ako getTask ne može da dohvati zadatak, tada se u finally bloku poziva metod processWorkerExit kojim nit izlazi.
Još jedan detalj: pošto Worker nasleđuje AQS, pre svakog izvršenja zadatka se poziva metod lock Workera, a nakon izvršenja poziva se unlock; time se preko stanja zaključavanja Workera može pročitati da li trenutna nit izvršava zadatak.
Ako želimo da proverimo da li nit izvršava zadatak, dovoljno je pozvati metod tryLock Workera — na osnovu toga da li je zaključavanje uspelo možemo proceniti; ako uspe, to znači da nit trenutno nije zaključana, odnosno ne izvršava zadatak; pri zatvaranju bazena niti metodom shutdown upravo se na taj način proverava da li nit izvršava zadatak — ako ne, pokušava se prekinuti one niti koje ne izvršavaju zadatak.
Pet: Kako nit dohvata zadatak i kako se ostvaruje istek vremena
Rekli smo da nakon izvršenja zadatka nastavlja da dohvata zadatke iz metode getTask, a ako ne uspe — izlazi. Pogledajmo implementaciju metode getTask.
private Runnable getTask() {
// indikator da li je poslednji poll() istekao
boolean timedOut = false;
// beskonačna petlja — sve dok se zadatak ne dohvati ili dok se ne odluči da radna nit izađe
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// ako je stanje bazena SHUTDOWN ili više (npr. STOP), a red zadataka je prazan, broj radnih niti se smanjuje i nit izlazi
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// proverava da li radna nit treba da bude završena nakon keepAliveTime bez izvršenja zadatka
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// ako broj radnih niti prelazi maksimalni, ili preovlađuje jezgreno i poslednji poll() je istekao, a red je prazan ili je broj niti veći od 1,
// pokušava se smanjiti broj radnih niti
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
// na osnovu indikatora timed bira se da li se čeka na zadatak neograničeno ili tokom keepAliveTime
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : // čeka se zadato vreme
workQueue.take(); // čeka se neograničeno
if (r != null) // zadatak uspešno dohvaćen
return r;
// ako je poll() istekao, postavlja se indikator timedOut
timedOut = true;
} catch (InterruptedException retry) {
// ako je nit prekinuta tokom čekanja, resetuje se timedOut i pokušava se ponovo
timedOut = false;
}
}
}Prethodni deo su provere stanja bazena niti; evo jednog ključnog reda:
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;Ovaj red služi da proveri da li nit koja dohvata zadatak sme da izađe zbog isteka vremena. Ako je allowCoreThreadTimeOut postavljeno na true ili je broj niti u bazenu veći od jezgrenog (corePoolSize), nit koja dohvata zadatak sme da izađe zbog isteka vremena.
Kako se ostvaruje izlazak zbog isteka vremena? Ovaj ključni red:
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();Na osnovu toga da li je istek dozvoljen, bira se metod poll ili metod take blokirajućeg reda workQueue. Ako je istek dozvoljen, poziva se metod poll sa argumentom keepAliveTime — vremenom neaktivnosti prosleđenim pri konstrukciji bazena niti; ovaj metod pokušava da blokira keepAliveTime vremena da dohvati zadatak iz reda i vraća null ako ne uspe; ako istek nije dozvoljen, poziva se metod take koji blokira dok god ne dohvati zadatak iz reda.
Odatle se jasno vidi kako se koristi keepAliveTime.
Dakle, sada bi trebalo da bude jasno zašto niti u bazenu mogu da izađu nakon određenog vremena neaktivnosti?
Suština je u tome što se iskorišćava metod poll blokirajućeg reda, koji dozvoljava zadato vreme čekanja; čim nit dostigne keepAliveTime bez dohvatanja zadatka, vraća se null, a čim getTask vrati null, nit izlazi.
Još jedan detalj: pri proveri da li sme da izađe zbog isteka vremena, ako je allowCoreThreadTimeOut postavljeno na true, sve niti u ovom redu imaju timed true, što znači da sve niti, uključujući i jezgrene, smeju da izađu zbog isteka vremena. Ako bazen niti želi da i jezgrene niti smeju da izađu zbog isteka, dovoljno je preko metode allowCoreThreadTimeOut postaviti tu promenljivu na true.
Tok cele metode getTask i mehanizma isteka niti je ilustrovan na slici:
Šest: 5 stanja bazena niti
Bazen niti interno ima 5 konstanti koje predstavljaju njegovih pet stanja:
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;- RUNNING: stanje u kojem se bazen nalazi odmah po kreiranju — može primati nove zadatke i obrađivati već dodate zadatke.
- SHUTDOWN: pozivom metode shutdown, bazen prelazi u stanje SHUTDOWN — tada više ne prima nove zadatke, ali nastavlja da obrađuje zadatke koji su već u redu.
- STOP: pozivom metode shutdownNow, bazen prelazi u stanje STOP — ne prima nove zadatke, ne nastavlja obradu zadataka iz reda, i pokušava da prekine niti koje trenutno obrađuju zadatke.
- TIDYING: u stanju SHUTDOWN, kada broj zadataka postane 0, odnosno kada su svi ostali zadaci završeni, bazen prelazi u TIDYING; ako je bazen u SHUTDOWN stanju i red zadataka je prazan i nema zadataka u obradi, bazen prelazi u TIDYING; ako je bazen u STOP stanju i nema zadataka u obradi, prelazi u TIDYING.
- TERMINATED: bazen je potpuno završen. Kada bazen u stanju TIDYING izvrši metod
terminated(), prelazi u stanje TERMINATED.
Stanje bazena se čuva u promenljivoj članu ctl; ctl ne čuva samo stanje bazena već i veličinu trenutnog broja niti u bazenu:
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));Na kraju, dijagram koji sumira tok ovih 5 stanja:
Zaista, tokom rada bazena niti, pre velike većine operacija se mora proveriti u kom stanju se bazen nalazi, pa tek onda odlučiti da li nastaviti sa tom operacijom.
Sedam: Gašenje bazena niti
Bazen niti pruža dva metoda za gašenje: shutdown i shutdownNow.
Metod shutdown
/**
* Pokreće uredno gašenje: u ovom gašenju izvršilac više ne prihvata nove zadatke,
* ali nastavlja da obrađuje postojeće zadatke iz reda.
* Kada se svi zadaci završe, niti u bazenu postepeno izlaze.
*/
public void shutdown() {
final ReentrantLock mainLock = this.mainLock; // glavna brava ThreadPoolExecutor-a
mainLock.lock(); // zaključavanje radi ekskluzivnog pristupa
try {
checkShutdownAccess(); // provera dozvole za gašenje
advanceRunState(SHUTDOWN); // ažuriranje stanja izvršioca na SHUTDOWN
interruptIdleWorkers(); // prekid svih neaktivnih radnih niti
onShutdown(); // hook metod u ScheduledThreadPoolExecutor-u, podklase ga mogu predefinisati radi dodatnih akcija
} finally {
mainLock.unlock(); // bez obzira na to kako try blok završi, brava se oslobađa
}
tryTerminate(); // ako uslovi dozvoljavaju, pokušava se završiti izvršilac
}Suština je u tome da se stanje bazena niti promeni u SHUTDOWN, a zatim pokuša prekid neaktivnih niti (kako se proverava neaktivnost, opisali smo kada smo pričali o Workeru koji nasleđuje AQS) — to su niti koje blokirano čekaju na zadatak.
Metod shutdownNow
/**
* Pokušava da zaustavi sve zadatke u toku, zaustavlja obradu zadataka koji čekaju,
* i vraća listu zadataka koji čekaju na obradu.
*
* @return lista zadataka koji nikada nisu počeli da se izvršavaju
*/
public List<Runnable> shutdownNow() {
List<Runnable> tasks; // lista za smeštanje neizvršenih zadataka
final ReentrantLock mainLock = this.mainLock; // glavna brava ThreadPoolExecutor-a
mainLock.lock(); // zaključavanje radi ekskluzivnog pristupa
try {
checkShutdownAccess(); // provera dozvole za gašenje
advanceRunState(STOP); // ažuriranje stanja izvršioca na STOP
interruptWorkers(); // prekid svih radnih niti
tasks = drainQueue(); // praznjenje reda i smeštanje rezultata u listu zadataka
} finally {
mainLock.unlock(); // bez obzira na to kako try blok završi, brava se oslobađa
}
tryTerminate(); // ako uslovi dozvoljavaju, pokušava se završiti izvršilac
return tasks; // vraća listu neizvršenih zadataka iz reda
}Suština je u tome da se stanje bazena promeni u STOP, pokuša prekid svih niti i uklanjanje preostalih zadataka iz blokirajućeg reda — upravo zato shutdownNow ne izvršava preostale zadatke.
Prema tome, glavna razlika između metoda shutdown i shutdownNow jeste u tome što posle shutdown niti i dalje mogu obrađivati zadatke iz reda, dok shutdownNow direktno uklanja zadatke iz reda, pa niti iz bazena više ne obrađuju te zadatke.
Osam: Nadzor bazena niti
Pri korišćenju bazena niti u projektu, obično je potrebno vršiti nadzor nad bazenom, kako bi se pri pojavi problema brzo locirao uzrok. Bazen niti pruža niz metoda za preuzimanje stanja rada.
- getCompletedTaskCount: broj zadataka koji su završili izvršenje
- getLargestPoolSize: najveći broj niti koje su ikada kreirane u bazenu. Ovo uglavnom služi da proverimo da li je bazen ikada bio pun.
- getActiveCount: preuzima broj niti koje trenutno izvršavaju zadatke
- getPoolSize: preuzima veličinu trenutnog broja niti u bazenu
Pored metoda koje bazen niti već implementira, on je rezervisao i niz metoda za proširenje. Na primer, u metodi runWorker se pre izvršenja zadatka poziva beforeExecute, a nakon izvršenja afterExecute; ti metodi su podrazumevano prazni — možete naslediti ThreadPoolExecutor i predefinisati ih kako biste ostvarili željenu funkcionalnost.
Devet: Scenariji upotrebe bazena niti
U Java programima se često javi potreba za višenitnom obradom nekog biznisa, ali se ne preporučuje kreiranje niti prostim nasleđivanjem klase Thread ili implementacijom interfejsa Runnable, jer to dovodi do učestalog kreiranja i uništavanja niti; s druge strane, kreiranje previše niti može izazvati rizik od iscrpljivanja resursa.
Zato je korišćenje bazena niti razumniji izbor — olakšava upravljanje zadacima i omogućava ponovno korišćenje niti. Bazen niti je opšteprihvaćen za scenarije koji zahtevaju asinhronu ili višenitnu obradu zadataka.
Evo nekoliko jednostavnih primera upotrebe bazena niti:
01: Simulacija veb servera:
Simuliramo jednostavan veb server koji prima zahteve i obrađuje ih preko bazena niti.
import java.util.concurrent.*;
public class SimpleWebServer {
private static final int NTHREADS = 100;
private static final ExecutorService exec = Executors.newFixedThreadPool(NTHREADS);
public static void main(String[] args) {
while (true) {
// prijem zahteva
Runnable request = new Runnable() {
public void run() {
// obrada zahteva
System.out.println("Request handled by " + Thread.currentThread().getName());
}
};
exec.execute(request);
}
}
}02: Paralelno računanje:
Korišćenje bazena niti za paralelno numeričko računanje.
import java.util.concurrent.*;
public class ParallelCalculation {
private static final int NTHREADS = 4;
private static final ExecutorService exec = Executors.newFixedThreadPool(NTHREADS);
public static void main(String[] args) {
Callable<Double> task = new Callable<Double>() {
@Override
public Double call() {
// ovde se simulira nešto numeričko računanje
return Math.random() * 100;
}
};
List<Future<Double>> results = new ArrayList<>();
for (int i = 0; i < 10; i++) {
results.add(exec.submit(task));
}
for (Future<Double> result : results) {
try {
System.out.println(result.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
exec.shutdown();
}
}03: Obrada asinhronih zadataka:
Simulacija obrade asinhronih zadataka.
import java.util.concurrent.*;
public class AsynchronousTaskProcessor {
private static final ExecutorService exec = Executors.newCachedThreadPool();
public static void main(String[] args) {
exec.execute(() -> {
// izvršenje nekog asinhronog zadatka
System.out.println("Async task started");
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Async task completed");
});
System.out.println("Main thread continues to execute other operations.");
exec.shutdown();
}
}Deset: Executors za izgradnju bazena niti i analiza problema
U gore navedenim primerima koristili smo klasu alata Executors koja dolazi u sastavu JDK-a da brzo kreiramo bazen niti.
- Bazen niti sa fiksnim brojem niti: broj jezgrenih niti je jednak maksimalnom broju niti
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}- Bazen niti sa jednom nit
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}- Bazen niti sa skoro neograničenim brojem niti
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}- Bazen niti sa mogućnošću vremenskog raspoređivanja
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}Iako JDK pruža brze metode za kreiranje bazena niti, ipak se ne preporučuje korišćenje Executors-a za tu svrhu — iz koda za konstrukciju bazena niti se vidi da newFixedThreadPool koristi LinkedBlockingQueue čiji je podrazumevani kapacitet reda neograničen, što u praksi pri previše zadataka može dovesti do prekoračenja memorije; newCachedThreadPool pak ima neograničen broj jezgrenih niti, pa pri previše zadataka može kreirati ogroman broj niti, što može dovesti do prevelikog opterećenja mašine i pada servisa.
To je inače jedno od čestih pitanja na intervjuima, pa obratite pažnju.
Jedanaest: Kako u realnim projektima razumno napraviti sopstveni bazen niti
Gore je pomenuto da bazeni niti kreirani preko alatke Executors zapravo ne mogu da zadovolje realne scenarije upotrebe; kako onda u realnom projektu konstruisati bazen niti i kako razumno podesiti parametre?
Broj niti
Podešavanje broja niti uglavnom zavisi od toga da li je biznis IO-intenzivan ili CPU-intenzivan.
CPU-intenzivan: podrazumeva zadatke koji uglavnom vrše obimna izračunavanja, bez mnogo toga što bi blokiralo nit. U takvim scenarijima broj niti se obično postavlja na broj CPU jezgara + 1.
IO-intenzivan: kada izvršenje zadatka zahteva mnogo U/I operacija, kao što su disk U/I, mrežni U/I, može doći do mnogo blokiranja, pa u IO-intenzivnim zadacima višenitnost može znatno ubrzati obradu. Broj niti se obično postavlja na 2 × broj CPU jezgara.
Metod u Javi za dobijanje broja CPU jezgara je: Runtime.getRuntime().availableProcessors();
Fabrika niti
Preporučuje se da napravite sopstvenu fabriku niti; pri kreiranju niti postavite ime niti, čime prilikom pregleda logova lako možete znati koja nit je izvršila kod.
Ograničeni red
Obično je potrebno podesiti veličinu ograničenog reda — na primer, LinkedBlockingQueue se pri konstrukciji može proslediti argument kojim se ograničava broj zadataka u redu, čime se izbegava da neprekidno dodavanje zadataka u red dovede do oom-a sistema.
Dobro, hajde da preko sopstvenog ThreadPoolExecutor-a preradimo prethodni primer sa Executors-om.
Simulacija veb servera:
Veb server obično mora da obrađuje U/I operacije, poput mrežnog U/I-a, pa se takvi zadaci posmatraju kao IO-intenzivni. Zato broj niti postavljamo na 2 × broj CPU jezgara.
import java.util.concurrent.*;
public class SimpleWebServer {
private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
private static final int CORE_POOL_SIZE = 2 * CPU_COUNT;
private static final int MAX_POOL_SIZE = 2 * CPU_COUNT + 1;
private static final ThreadPoolExecutor exec = new ThreadPoolExecutor(
CORE_POOL_SIZE,
MAX_POOL_SIZE,
60L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(1000)
);
public static void main(String[] args) {
while (true) {
Runnable request = () -> System.out.println("Request handled by " + Thread.currentThread().getName());
exec.execute(request);
}
}
}Paralelno računanje:
Paralelni zadaci se uglavnom koriste za računanje, bez U/I blokiranja, pa su CPU-intenzivni. Broj niti se postavlja na broj CPU jezgara + 1.
import java.util.*;
import java.util.concurrent.*;
public class ParallelCalculation {
private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
private static final int CORE_POOL_SIZE = CPU_COUNT + 1;
private static final int MAX_POOL_SIZE = CPU_COUNT * 2;
private static final ThreadPoolExecutor exec = new ThreadPoolExecutor(
CORE_POOL_SIZE,
MAX_POOL_SIZE,
10L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(1000)
);
public static void main(String[] args) {
Callable<Double> task = () -> Math.random() * 100;
List<Future<Double>> results = new ArrayList<>();
for (int i = 0; i < 10; i++) {
results.add(exec.submit(task));
}
for (Future<Double> result : results) {
try {
System.out.println(result.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
exec.shutdown();
}
}Obrada asinhronih zadataka:
Asinhroni zadaci obično uključuju U/I operacije, kao što su upiti u bazu ili čitanje/pisanje datoteka, pa se takvi zadaci posmatraju kao IO-intenzivni. Broj niti se postavlja na 2 × broj CPU jezgara.
import java.util.concurrent.*;
public class AsynchronousTaskProcessor {
private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
private static final int CORE_POOL_SIZE = 2 * CPU_COUNT;
private static final int MAX_POOL_SIZE = 2 * CPU_COUNT + 2;
private static final ThreadPoolExecutor exec = new ThreadPoolExecutor(
CORE_POOL_SIZE,
MAX_POOL_SIZE,
60L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(1000)
);
public static void main(String[] args) {
exec.execute(() -> {
System.out.println("Async task started");
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Async task completed");
});
System.out.println("Main thread continues to execute other operations.");
exec.shutdown();
}
}Dvanaest: Kratak pregled
Ovaj članak je uglavnom predstavio princip rada bazena niti i scenarije upotrebe; bazen niti se uglavnom ostvaruje preko blokirajućeg reda, a scenariji upotrebe su asinhrona ili višenitna obrada zadataka. Bazen niti se može brzo kreirati preko Executors-a, ali se to ne preporučuje jer bazeni kreirani preko Executors-a imaju izvesnih mana — na primer, neograničeni red može dovesti do prekoračenja memorije, a neograničen broj niti može dovesti do prevelikog opterećenja mašine. Zato se u realnim projektima preporučuje izgradnja sopstvenog bazena niti ThreadPoolExecutor i razumno podešavanje parametara — broja niti, veličine reda itd. — prema biznis scenariju.
Autor: Chenmo Wang Er. Deo sadržaja potiče iz članka na WeChat javnom nalogu čitaoca San You, vrlo dobro napisan — toplo preporučujemo.
lajkuj i prati
GitHub otvorena baza znanja sa 17000+ zvezdica „Ergov put napredovanja u Javi"; druga knjiga u PDF-u „Knjiga o konkurentnom programiranju" je konačno stigla! Obuhvata osnovne koncepte i načine korišćenja niti, Java memorijski model, synchronized, volatile, CAS, AQS, ReentrantLock, bazen niti, konkurentne kontejnere, ThreadLocal, model proizvođač-potrošač i druge teme koje su obavezne za intervju i razvoj — ukupno preko 150 hiljada reči i 200+ ručno crtanih ilustracija, pristupačno i sa humorom... više detalja na: Ergova knjiga o konkurentnom programiranju.pdf.