Detaljna analiza Java zakazanih zadataka ScheduledThreadPoolExecutor
Klasa ScheduledThreadPoolExecutor za zakazane zadatke ima dve svrhe: izvršavanje zadatka nakon određenog vremenskog kašnjenja i periodično ponavljano izvršavanje zadatka.
Pre JDK 1.5 za zakazane zadatke uglavnom se koristila klasa Timer, ali Timer ima sledeće nedostatke:
- Timer radi u jednonitnom režimu;
- Ako neki TimerTask dugo traje tokom izvršavanja, uticaće na raspoređivanje drugih zadataka;
- Raspoređivanje zadataka u Timer-u zasnovano je na apsolutnom vremenu, pa je osetljivo na sistemsko vreme;
- Timer ne hvata izuzetke bačene pri izvršavanju TimerTask-a; pošto je Timer jednonitan, jednom kad se pojavi izuzetak nit se prekida i drugi zadaci ne mogu da se izvrše.
Zato su od JDK 1.5 programeri napustili Timer i prešli na ScheduledThreadPoolExecutor. Prvo, pogledajte sliku da biste stekli utisak.
Primer upotrebe
Pretpostavimo da imamo sledeći zahtev: u određeno vreme poslati poruku drugima. Poruku (uključujući vreme slanja) čuvamo u bazi, a zatim zakazanim zadatkom svake sekunde proveravamo u bazi ima li poruka koje treba poslati u tekućem trenutku. Kako se taj zadatak ostvaruje? Ispod je demo:
public class ThreadPool {
private static final ScheduledExecutorService executor = new
ScheduledThreadPoolExecutor(1, Executors.defaultThreadFactory());
private static SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
public static void main(String[] args){
// Novi zakazani zadataka sa fiksnim kašnjenjem
executor.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
if (haveMsgAtCurrentTime()) {
System.out.println(df.format(new Date()));
System.out.println("Pažnja svima, šaljem poruku");
}
}
}, 1, 1, TimeUnit.SECONDS);
}
public static boolean haveMsgAtCurrentTime(){
//upit u bazi: ima li poruka koje treba poslati u tekućem trenutku
//ovde se implementacija izostavlja i vraća se true
return true;
}
}Ispod je deo ispisa (demo će se izvršavati neprekidno):
2023-08-24 16:16:48
Pažnja svima, šaljem poruku
2023-08-24 16:16:49
Pažnja svima, šaljem poruku
2023-08-24 16:16:50
Pažnja svima, šaljem poruku
2023-08-24 16:16:51
Pažnja svima, šaljem poruku
2023-08-24 16:16:52
Pažnja svima, šaljem poruku
2023-08-24 16:16:53
Pažnja svima, šaljem poruku
2023-08-24 16:16:54
Pažnja svima, šaljem poruku
2023-08-24 16:16:55
Pažnja svima, šaljem porukuTo je jedna jednostavna primena ScheduledThreadPoolExecutor-a; sledeće ćemo se upustiti u principe realizacije.
Struktura klase
public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor
implements ScheduledExecutorService {
public ScheduledThreadPoolExecutor(int corePoolSize,ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory);
}
//……
}ScheduledThreadPoolExecutor nasleđuje ThreadPoolExecutor i implementira interfejs ScheduledExecutorService. Bazen niti ThreadPoolExecutor je već bio opisan; nadamo se da se još sećate. Sledeće pogledajmo interfejs ScheduledExecutorService.
public interface ScheduledExecutorService extends ExecutorService {
/**
* Zakazuje izvršavanje Runnable zadatka nakon zadatog kašnjenja.
*
* @param command zadatak koji treba izvršiti
* @param delay vreme kašnjenja
* @param unit jedinica vremena
* @return ScheduledFuture pomoću kojeg se može dobiti rezultat ili otkazati zadatak
*/
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
/**
* Zakazuje izvršavanje Callable zadatka nakon zadatog kašnjenja.
*
* @param callable zadatak koji treba izvršiti
* @param delay vreme kašnjenja
* @param unit jedinica vremena
* @return ScheduledFuture pomoću kojeg se može dobiti rezultat ili otkazati zadatak
*/
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
/**
* Zakazuje izvršavanje Runnable zadatka: prvi put nakon inicijalnog kašnjenja, a zatim svaki period vremena.
*
* @param command zadatak koji treba izvršiti
* @param initialDelay inicijalno kašnjenje prvog izvršavanja
* @param period vremenski razmak između uzastopnih izvršavanja
* @param unit jedinica vremena
* @return ScheduledFuture pomoću kojeg se može dobiti rezultat ili otkazati zadatak
*/
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
/**
* Zakazuje izvršavanje Runnable zadatka: prvi put nakon inicijalnog kašnjenja, a zatim nakon svakog završetka zadatka čeka zadato kašnjenje i ponavlja izvršavanje.
*
* @param command zadatak koji treba izvršiti
* @param initialDelay inicijalno kašnjenje prvog izvršavanja
* @param delay kašnjenje nakon svakog završetka izvršavanja
* @param unit jedinica vremena
* @return ScheduledFuture pomoću kojeg se može dobiti rezultat ili otkazati zadatak
*/
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);
}Interfejs ScheduledExecutorService nasleđuje interfejs ExecutorService i dodaje nekoliko metoda vezanih za zakazivanje. Prve dve metode služe za jednokratno zakazano izvršavanje zadatka; razlika je u tome što jedna ima povratnu vrednost, a druga nema.
Detaljnije ćemo opisati sledeće dve metode:
01,scheduleAtFixedRate
Metoda scheduleAtFixedRate prvi put izvršava zadatak nakon initialDelay, a zatim ga ponovo izvršava svaki period. Pazite, period se računa od početka izvršavanja zadatka. Nakon početka izvršavanja tajmer svakih period proverava da li je zadatak završen; ako jeste, ponovo ga pokreće, a ako nije, čeka da se zadatak završi i tek ga onda pokreće. Pogledajte sliku:
02,scheduleWithFixDelay
Ova metoda prvi put izvršava zadatak nakon initialDelay, a zatim, pošto se zadatak završi, čeka delay i ponovo ga izvršava. Pogledajte sliku.
Sigurni smo da razliku možete da osetite.
Glavne metode
schedule
// Zadatak command izvršava se nakon delay; ovaj zadatak se izvršava samo jednom
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
// Ovde decorateTask samo vraća drugi argument
RunnableScheduledFuture<?> t = decorateTask(command,
new ScheduledFutureTask<Void>(command, null, triggerTime(delay,unit)));
// Glavna metoda za zadataka sa kašnjenjem ili periodične zadatke; biće objašnjena kasnije
delayedExecute(t);
return t;
}Prvo pogledajmo grafikon odnosa nekoliko klasa i interfejsa koji se ovde pojavljuju:
Interfejs Delayed
// Nasleđuje interfejs Comparable, što znači da objekti klase podržavaju sortiranje
public interface Delayed extends Comparable<Delayed> {
// Vraća preostalo kašnjenje ovog objekta
long getDelay(TimeUnit unit);
}Interfejs Delayed je vrlo jednostavan: nasleđuje interfejs Comparable, što znači da objekti mogu biti upoređivani i sortirani.
Interfejs ScheduledFuture
// Samo nasleđuje interfejse Delayed i Future, bez ikakvog sopstvenog koda
public interface ScheduledFuture<V> extends Delayed, Future<V> {
}Interfejs RunnableScheduledFuture
public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> {
// Da li je u pitanju periodičan zadatak; periodičan zadatak se može rasporediti na višestruko izvršavanje, a neperiodičan samo jednom
boolean isPeriodic();
}Klasa ScheduledFutureTask
Vratimo se metodi schedule; ona kreira objekat ScheduledFutureTask; iz gornjeg grafikona se vidi da ScheduledFutureTask direktno ili indirektno implementira mnoge interfejse. Pogledajmo implementacije unutar ScheduledFutureTask.
Konstruktor
ScheduledFutureTask(Runnable r, V result, long ns, long period) {
// Poziv konstruktora roditeljske klase FutureTask
super(r, result);
// time predstavlja vreme sledećeg izvršavanja zadatka
this.time = ns;
// Kod periodičnog zadatka, pozitivna vrednost označava fiksnu stopu, negativna fiksno kašnjenje, 0 da nije periodičan
this.period = period;
// Broj zadatka
this.sequenceNumber = sequencer.getAndIncrement();
}Implementacija interfejsa Delayed
// Implementacija metode getDelay interfejsa Delayed; vraća preostalo vreme do početka izvršavanja zadatka
public long getDelay(TimeUnit unit) {
return unit.convert(time - now(), TimeUnit.NANOSECONDS);
}Implementacija interfejsa Comparable
// Metod compareTo interfejsa Comparable — poredi „veličinu" dva zadatka.
public int compareTo(Delayed other) {
if (other == this)
return 0;
if (other instanceof ScheduledFutureTask) {
ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
long diff = time - x.time;
// Manje od 0: trenutni zadatak se izvršava pre other-a, pa u redu sa kašnjenjem ide ispred other-a
if (diff < 0)
return -1;
// Veće od 0: trenutni zadatak se izvršava posle other-a, pa u redu sa kašnjenjem ide iza other-a
else if (diff > 0)
return 1;
// Ako su vremena izvršavanja ista, upoređuju se brojevi zadataka; manji broj ide ispred, veći iza
else if (sequenceNumber < x.sequenceNumber)
return -1;
else
return 1;
}
// Ako tip zadatka nije ScheduledFutureTask, poredi se preko metode getDelay
long d = (getDelay(TimeUnit.NANOSECONDS) -
other.getDelay(TimeUnit.NANOSECONDS));
return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
}setNextRunTime
// Nakon izvršavanja zadatka postavlja se vreme sledećeg izvršavanja
private void setNextRunTime() {
long p = period;
// p > 0: zadatak koji se izvršava fiksnom stopom
// Dovoljno je na vreme početka prethodnog izvršavanja dodati p
if (p > 0)
time += p;
// p < 0: zadatak koji se izvršava sa fiksnim kašnjenjem
// Sledeće izvršavanje je u trenutnom vremenu (vreme završetka zadatka) plus vreme -p
else
time = triggerTime(-p);
}Implementacija interfejsa Runnable
public void run() {
boolean periodic = isPeriodic();
// Ako tekuće stanje ne dozvoljava izvršavanje zadatka, otkaže ga
if (!canRunInCurrentRunState(periodic))
cancel(false);
// Nije periodičan zadatak — izvrši ga jednom, pozivom run metode roditeljske klase
else if (!periodic)
ScheduledFutureTask.super.run();
// Periodičan zadatak — poziva runAndReset metodu FutureTask-a; po završetku
// ponovo postavlja vreme sledećeg izvršavanja i ponovo stavlja zadatak u red, čekajući sledeće zakazivanje
else if (ScheduledFutureTask.super.runAndReset()) {
setNextRunTime();
reExecutePeriodic(outerTask);
}
}Sumirajmo tok izvršavanja metode run:
- Ako tekuće stanje bazena niti ne dozvoljava izvršavanje zadataka, zadatak se otkaže i odmah se vraća; inače, pređite na korak 2;
- Ako nije periodičan zadatak, poziva se run metoda iz FutureTask-a, postavlja se rezultat izvršavanja i odmah se vraća; inače, pređite na korak 3;
- Ako je periodičan zadatak, poziva se runAndReset metoda iz FutureTask-a, bez postavljanja rezultata; ako uspe, pređite na korake 4 i 5;
- Izračunava se konkretno vreme sledećeg izvršavanja zadatka;
- Zadatak se ponavlja.
Metod runAndReset projektovan je za višestruko izvršavanje zadatka. Nakon izvršavanja, runAndReset ne postavlja rezultat izvršavanja, ne ažurira stanje zadatka, već održava stanje zadatka u inicijalnom (NEW) stanju — to je razlika u odnosu na run metodu klase FutureTask.
scheduleAtFixedRate
Pogledajmo kod:
// Pazite, i fiksna stopa i fiksno kašnjenje primaju Runnable, što znači da ovi zakazani zadaci nemaju povratnu vrednost
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
if (period <= 0)
throw new IllegalArgumentException();
// Kreira zadatak sa inicijalnim kašnjenjem i fiksnim periodom
ScheduledFutureTask<Void> sft =
new ScheduledFutureTask<Void>(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(period));
RunnableScheduledFuture<Void> t = decorateTask(command, sft);
// outerTask predstavlja zadatak koji će se ponovo dodati u red
sft.outerTask = t;
// Biće objašnjeno kasnije
delayedExecute(t);
return t;
}Metoda scheduleAtFixedRate slična je metodi schedule; razlika je u tome što scheduleAtFixedRate interno kreira ScheduledFutureTask sa inicijalnim kašnjenjem i fiksnim periodom.
scheduleWithFixedDelay
Ni scheduleWithFixedDelay se razlikuje samo po tome što kreira drugačiji ScheduledFutureTask.
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
if (delay <= 0)
throw new IllegalArgumentException();
// Kreira zadatak sa inicijalnim kašnjenjem i fiksnim kašnjenjem
ScheduledFutureTask<Void> sft =
new ScheduledFutureTask<Void>(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(-delay));
RunnableScheduledFuture<Void> t = decorateTask(command, sft);
// outerTask predstavlja zadatak koji će se ponovo dodati u red
sft.outerTask = t;
// Biće objašnjeno kasnije
delayedExecute(t);
return t;
}delayedExecute
I schedule, i scheduleAtFixedRate, i scheduleWithFixedDelay se naposletku pozivaju na metodu delayedExecute, koja je glavna metoda izvršavanja zakazanih zadataka. Pogledajmo izvorni kod:
private void delayedExecute(RunnableScheduledFuture<?> task) {
// Ako je bazen niti zatvoren, poziva se procesor za odbijeno izvršavanje
if (isShutdown())
reject(task);
else {
// Zadatak se dodaje u red čekanja
super.getQueue().add(task);
// Ako je bazen niti zatvoren, a tekuće stanje ne dozvoljava izvršavanje ovog zadatka, zadatak se uklanja iz reda i otkazuje
if (isShutdown() &&
!canRunInCurrentRunState(task.isPeriodic()) &&
remove(task))
task.cancel(false);
else
// Dodaje se worker; čak i ako je corePoolSize=0, dodaje se worker
ensurePrestart();
}
}Logika delayedExecute je jednostavna: dodaje zadatak u red čekanja, a zatim poziva metodu ensurePrestart.
void ensurePrestart() {
int wc = workerCountOf(ctl.get());
if (wc < corePoolSize)
addWorker(null, true);
else if (wc == 0)
addWorker(null, false);
}Metoda ensurePrestart uglavnom poziva metodu addWorker; u bazenu niti se radne niti pokreću i zadaci izvršavaju preko te metode. Nadamo se da se još sećate.
Za ScheduledThreadPoolExecutor, nakon što se worker doda u bazen niti, u redu čekanja čeka da dobije zadatak, što je u skladu sa ThreadPoolExecutor-om. Ali kako worker uzima zakazane zadatke iz reda čekanja?
DelayedWorkQueue
ScheduledThreadPoolExecutor koristi DelayedWorkQueue da čuva zadatke u čekanju.
Na vrhu tog reda trebalo bi da bude zadatak koji će se uskoro izvršiti, pa worker brine samo o zadatku na vrhu; ako vreme početka izvršavanja zadatka na vrhu još nije stiglo, worker treba da nastavi da čeka.
DelayedWorkQueue je neograničen prioritetni red, koji za skladištenje koristi niz, a na donjem nivou prioritetni red ostvaruje preko strukture gomile.
Može se predstaviti sledećim nizom:
U toj strukturi mogu se uočiti sledeće osobine. Pretpostavimo da indeksi počinju od 0; indeks deteta je k, a indeks roditelja p:
- Indeks levog deteta čvora: k = p * 2 + 1;
- Indks desnog deteta čvora: k = (p + 1) * 2;
- Indeks roditelja čvora: p = (k - 1) / 2.
Pogledajmo deklaraciju i članove DelayedWorkQueue:
static class DelayedWorkQueue extends AbstractQueue<Runnable>
implements BlockingQueue<Runnable> {
// Inicijalni kapacitet reda
private static final int INITIAL_CAPACITY = 16;
// Niz za čuvanje zakazanih zadataka; sortiranje gomile ostvaruje se preko niza
private RunnableScheduledFuture[] queue = new RunnableScheduledFuture[INITIAL_CAPACITY];
// Nit koja trenutno čeka na vrhu reda
private Thread leader = null;
// Brava i nadgledač za leader nit
private final ReentrantLock lock = new ReentrantLock();
private final Condition available = lock.newCondition();
// ostali kod, izostavljeno
}Kada nit postane leader, čeka samo vreme kašnjenja zadatka sa vrha reda; ostale niti čekaju bezuslovno. Leader pre nego što vrati zadatak mora da obavesti druge niti, sve dok neka nit ne postane novi leader. Kad god zadatak na vrhu bude zamenjen drugim koji se ranije izvršava, leader se postavlja na null, a druge niti na čekanju (koje je trenutni leader obavestio) i trenutni leader ponovo se takmiče za leadera.
Sve niti imaju jednu od tri uloge: leader, follower, te jednu radnu ulogu: proccesser. Osnovno načelo je da istovremeno najviše jedna nit bude leader. Svi follower-i čekaju da postanu leader. Pri pokretanju bazena niti automatski se stvara Leader koji čeka mrežne IO događaje; kada se događaj pojavi, Leader najpre obaveštava jednog Follower-a koji postaje novi Leader, a zatim sam odlazi da radi — da obradi taj mrežni događaj; po završetku priključuje se redu Follower niti i čeka sledeći put da postane Leader. Ovakav pristup poboljšava srodnost CPU keša i eliminiše dinamičko zauzeće memorije i razmenu podataka među nitima.
Istovremeno, definišu se brava ReentrantLock lock i Condition available, koji služe za upravljanje i obaveštavanje sledeće niti pri takmičenju za leadera.
Kada novi zadatak postane prvi u redu ili kada je potrebno da nova nit postane leader, niti na available nadgledaču se obaveštavaju i takmiče da postanu leader nit, što je pomalo slično obrascu proizvođač-potrošač.
DelayedWorkQueue je prioritetni red koji garantuje da će zadatak koji izađe iz reda uvek biti onaj sa najblažim vremenom izvršavanja među trenutnim zadacima u redu; pošto je zasnovan na strukturi gomile, najgora vremenska složenost umetanja i brisanja je O(logN).
Sada pogledajmo nekoliko važnijih metoda u DelayedWorkQueue.
take
public RunnableScheduledFuture take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
// Uzmi zadatak sa vrha gomile — to je zadatak koji se najskorije izvršava
RunnableScheduledFuture first = queue[0];
// Vrh gomile je prazan — nit mora čekati na uslovu available
if (first == null)
available.await();
else {
// Koliko vremena preostaje do izvršavanja zadatka sa vrha
long delay = first.getDelay(TimeUnit.NANOSECONDS);
// Zadatak sa vrha već može da se izvrši; finishPoll ponovo preuređuje gomilu kako bi zadovoljila osobine minimalne gomile; metoda postavlja indeks zadatka u
// gomili na -1 i vraća taj zadatak
if (delay <= 0)
return finishPoll(first);
// Ako leader nije null, to znači da je već jedna nit postala leader i čeka da vreme
// izvršavanja zadatka sa vrha nastupi; u tom slučaju druge niti moraju čekati na uslovu available
else if (leader != null)
available.await();
else {
// leader je null — trenutna nit postaje novi leader
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
// Pošto je trenutna nit postala leader, treba samo da sačeka vreme izvršavanja zadatka sa vrha
available.awaitNanos(delay);
} finally {
// Pre vraćanja elementa sa vrha, leader se postavlja na null
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
// Obaveštava ostale niti koje čekaju na uslovu available; one mogu da se takmiče za novog leadera
if (leader == null && queue[0] != null)
available.signal();
lock.unlock();
}
}Kada se poziva take?
U tekstu o bazenu niti opisivali smo metodu getTask; radne niti u petlji uzimaju zadatke iz workQueue. Ali kod zakazanih zadataka je drugačije: jednom kad metoda getTask uzme zadatak, on počinje da se izvršava, a možda vreme izvršavanja još nije nastupilo. Zato metoda take mora da osigura da zadatak može biti preuzet tek u zadato vreme izvršavanja.
Sumirajmo tok:
- Ako element sa vrha gomile ne postoji, čeka se na available.
- Ako je vreme izvršavanja zadatka sa vrha nastupilo, element sa vrha se zamenjuje poslednjim elementom gomile i gomila se preuređuje tako da zadovoljava osobine minimalne gomile; istovremeno se indeks zadatka u gomili postavlja na -1 i zadatak se vraća.
- Ako leader nije null, to znači da je već jedna nit postala leader; ostale niti moraju čekati na available nadgledaču.
- Ako je leader null, trenutna nit postaje novi leader i čeka do vremena izvršavanja zadatka sa vrha.
- Pre povratka iz metode take, leader se postavlja na null i obaveštavaju se ostale niti.
Još o ulozi leader-a: ovde leader služi da se smanji nepotrebno vremensko čekanje — kad nit postane leader, čeka samo vremenski interval sledećeg čvora, dok druge niti čekaju neograničeno. Leader nit pre povratka iz take() ili poll() mora da signalizira druge niti, osim ako je neka druga nit postala leader.
Na primer, bez leader-a, pri izvršavanju take uvek bi se izvršilo available.awaitNanos(delay). Pretpostavimo da je trenutna nit izvršila taj kod, a da signal još nije stigao; ako i druga nit izvrši isti kod, i ona bi bila blokirana.
Ali samo jedna nit vraća zadatak sa vrha reda; ostale niti posle awaitNanos(delay) nastavljaju for petlju; pošto je zadatak sa vrha već vraćen, novi zadatak sa vrha je drugi, pa treba ponovo proceniti vreme i ponovo blokirati.
Da se više niti ne bi često i uzalud vremenski čekale, uveden je leader; ako leader nije null, to znači da prvi čvor u redu već čeka da izađe, pa u tom trenutku ostale niti stalno čekaju, čime se smanjuje nepotrebno čekanje (pazite, u finally se poziva signal() za buđenje jedne niti, a ne signalAll()).
offer
Ova metoda ubacuje vrednost u red i vraća da li je ubacivanje uspelo.
public boolean offer(Runnable x) {
if (x == null)
throw new NullPointerException();
RunnableScheduledFuture e = (RunnableScheduledFuture)x;
final ReentrantLock lock = this.lock;
lock.lock();
try {
int i = size;
// Ako je broj elemenata u redu veći ili jednak dužini niza, potrebno je proširenje; novi kapacitet gomile je 1,5 puta veći od starog
if (i >= queue.length)
grow();
// Broj elemenata u gomili se povećava za 1
size = i + 1;
// Preuređivanje gomile
if (i == 0) {
queue[0] = e;
setIndex(e, 0);
} else {
// Preuređivanje gomile kako bi zadovoljila minimalnu gomilu; poređenje se vrši gore pomenutim compareTo metodom
siftUp(i, e);
}
if (queue[0] == e) {
leader = null;
// Obaveštavanje ostalih niti koje čekaju na uslovu available; mogu se takmičiti za novog leadera
available.signal();
}
} finally {
lock.unlock();
}
return true;
}Metod offer realizuje ubacivanje zadatka u red sa kašnjenjem i osigurava da čitav red i dalje zadovoljava osobine minimalne gomile.
Minimalna gomila (Min Heap) je potpuno binarno stablo u kojem je vrednost svakog roditeljskog čvora manja ili jednaka vrednostima njegove dece. Drugim rečima, u minimalnoj gomili koren (vrh stabla) sadrži najmanju vrednost među svim čvorovima.
I minimalnu gomilu smo već pominjali. Pogledajmo metodu siftUp koja služi za preuređivanje gomile.
private void siftUp(int k, RunnableScheduledFuture<?> key) {
while (k > 0) {
// Pronađi indeks roditeljskog čvora
int parent = (k - 1) >>> 1;
// Dohvati roditeljski čvor
RunnableScheduledFuture<?> e = queue[parent];
// Ako je vreme izvršavanja čvora key veće od vremena roditeljskog čvora, nema potrebe za daljim sortiranjem
if (key.compareTo(e) >= 0)
break;
// Ako je key.compareTo(e) < 0, to znači da je vreme izvršavanja čvora key manje od vremena roditeljskog čvora; roditeljski čvor se pomera unazad
queue[k] = e;
// Postavi indeks na k
setIndex(e, k);
k = parent;
}
// key se postavlja na poziciju nakon sortiranja
queue[k] = key;
setIndex(key, k);
}Kod je lako razumeti: u petlji se, na osnovu čvora key i njegovog roditelja, procenjuje — ako je vreme izvršavanja čvora key manje od vremena roditeljskog čvora, dva čvora se zamene, tako da čvor sa ranijim vremenom izvršavanja ide ispred u redu.
Pretpostavimo da je vreme kašnjenja novog čvora (dobijeno pozivom getDelay()) 5; proces izgleda ovako:
- Najpre se novi čvor dodaje na kraj niza; tada je indeks novog čvora k jednak 7:
- Izračunava se indeks novog roditeljskog čvora:
parent = (k - 1) >>> 1, parent = 3; tada je vremenski intervalqueue[3]jednak 8; pošto je5 < 8, izvršava sequeue[7] = queue[3]:
- Sada se k postavlja na 3, nastavlja se petlja, ponovo se izračunava parent kao 1; vremenski interval
queue[1]je 3; pošto je5 > 3, izlazi se iz petlje, konačno k je 3:
Vidi se da pri svakom dodavanju čvora procena zavisi samo od roditeljskog čvora, a ne utiče na braću/sestre.
Zaključak
ScheduledThreadPoolExecutor je bazen niti za zakazane zadatke; njegova glavna uloga je periodično izvršavanje zadataka. Princip realizacije mu se oslanja na DelayedWorkQueue za čuvanje zadataka u čekanju; DelayedWorkQueue je neograničen prioritetni red, koji za skladištenje koristi niz, a na donjem nivou prioritetni red ostvaruje preko strukture gomile.
Urednik: Chenmo Wang Er; originalni sadržaj potiče iz open-source repozitorijuma koji je prijatelj Xiao Qi Yinghuochong objavio: Duboko i pristupačno o Java višenitnosti, toplo preporučujemo. Ostali reference:
Preporučeno čitanje: 11 načina realizacije zadataka sa kašnjenjem čitaoca San You