Všeobecné info

• java.util.concurrent.ThreadPoolExecutor
• univerzálny vykonávateľ úloh Runnable / Callable
• implementuje ExecutorService
• úlohy sa submitujú cez submit() alebo cez execute()

Konštrukcia

• možno využiť statické metódy z Executors.
• tie v skutočnosti volajú konštruktor

Priame vytváranie

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) 

• corePoolSize: počet vlákien, ktoré ostanú v poole, aj keď nemajú žiadne úlohy na vykonávanie
• maximumPoolSize: maximálny počet vlákien v poole
• keepAliveTime: ako dlho má vyčkávať idle vlákno na nové úlohy? Po vypršaní sa vlákno ukončí.
• unit: časová jednotka pre keepAliveTime
• workQueue: implementácia frontu pre úlohy, ktoré čakajú na spracovanie vo vláknach
• threadFactory: voliteľný parameter. Továreň pre nové vlákna.
• handler: voliteľný argument. Politika pre úlohy, ktoré sú zamietnuté pre nedostatok voľných vlákien / preplnený front čakajúcich vlákien.

Správanie pri submitovaných úlohách

• beží menej než corePoolSize vlákien?
  • pridávanie vlákien má vždy prednosť pred radením do frontu
  • vždy sa vytvorí nové vlákno
  • i vtedy, keď sú ostatné vlákna v poole v pokoji (idle)
• beží >= corePoolSize a < maximumPoolSize vlákien?
  • radenie do frontu má vždy prednosť pred pridávaním vlákien
  • úloha sa zaradí do frontu
  • ak ju nemožno zaradiť (napr. front je plný), vytvorí sa nové vlákno
    • front odmieta, ak offer() vráti false
• beží >= maximumPoolSize vlákien?
  • úloha je zamietnutá
  • ďalšie správanie podľa politiky zamietania

Špeciálna konfigurácia poolu

• corePoolSize == maximumPoolSize?
  • pool pevnej veľkosťi
• maximumPoolSize == Integer.MAX_VALUE?
  • front nikdy nebude plný
  • => vždy sa vytvorí nové vlákno

Vytváranie zo statických metód

newFixedThreadPool(int nThreads)

Bežíme na fixnom počte vlákien.

new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())

• corePoolSize == maximumPoolSize: nThreads
• keepAliveTime: 0 ms
• workQueue: neohraničený front LinkedBlockingQueue

newCachedThreadPool()

Vlákna vznikajú podľa potreby, a idle vlákna sa recyklujú.

new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>())

• corePoolSize: 0
• maximumPoolSize: neohraničený
• keepAliveTime: 1 minúta
• workQueue: úlohy sa priamo odovzdávajú vláknam cez SynchronousQueue:

newSingleThreadExecutor()

Bežíme na jedinom vlákne. Prakticky fixed thread pool pre n=1.

new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())

• corePoolSize == maximumPoolSize: 1. Bežíme na jedinom vlákne.
• keepAliveTime: 0 milisekúnd
• workQueue: neohraničený front LinkedBlockingQueue

ThreadPoolExecutor a radenie do frontu

Priame odovzdávanie

• žiadny front = využije sa SynchronousQueue
• vždy sa vytvorí nové vlákno
  • radenie do frontu je zakázané
• tradične sa nastaví neohraničený maximumPoolSize
  • hranica by spôsobila zamietanie úloh
• ak úlohy prichádzajú rýchlejšie než sa spracovávajú
  • exploduje počet vlákien
• zabraňuje sa problémom pri úlohách, ktoré sú na sebe závislé

SynchronousQueue

• blokujúci front
• každé vloženie musí čakať na odobratie z iného vlákna
  • a naopak
• front nemá internú kapacitu!
  • tvári sa ako prázdna kolekcia
• nefunguje peek():
  • element je prítomný len pri odobratí
• nefunguje vkladanie, pokiaľ ho iné vlákno nechce odobrať
• nefunguje iterovanie:
  • nie je čo iterovať
• hlava head je element, ktorý bol zaradený do frontu
  • ak nebolo zaradené nič, poll() vráti null
• podobné rendezvous kanálom z CSP/Ada

Neohraničené fronty

• ak má bežať najviac corePoolSize vlákien
• ostatné úlohy sa radia do frontu
  • napr. LinkedBlockingQueue bez kapacity
• maximumPoolSize teda nemá efekt
• vhodné pre prípady, kde sa úlohy neovplyvňujú
• front je priehrada pre záplavy úloh
• ak úlohy prichádzajú rýchlejšie než sa spracovávajú
  • exploduje veľkosť frontu

Ohraničené fronty

• zabraňujú vyčerpaniu zdrojov
• maximumPoolSize aj veľkosť frontu sú ohraničené
• možno vylaďovať oba parametre
• veľké fronty a malé pooly: nízky prietok, šetrenie CPU/zdrojov OS/prepínania úloh
• malé fronty a veľké pooly: vyťaženejšie CPU, ale pozor na prepínanie úloh

Politika pre zamietnuté úlohy

• zamietnutá úloha:
  • exekútor sa vypína (po zavolaní shutdown())
  • vyčerpal sa počet dostupných vlákien
  • a vyčerpal sa front úloh
• politiky:
  • AbortPolicy: vyhodí sa výnimka RejectedExecutionException
  • CallerRunsPolicy: úloha sa spustí vo vlákne, ktoré zavolalo execute(). Prakticky sa očividne spomalí zasielanie úloh.
  • DiscardPolicy: úloha sa zahodí
  • DiscardOldestPolicy: úloha na začiatku frontu sa zahodí, vykoná sa opakované vykonanie úlohy (podľa potreby viackrát).