streamalism.org

1. Úvod

Tento článok pojednáva o skupinách vlákien v prostredí Java - počnúc rôznymi implementáciami v štandardnej knižnici Java a potom pohľadom na knižnicu Guava spoločnosti Google.

2. Pool vlákien

V prostredí Java sú vlákna mapované na vlákna na úrovni systému, ktoré sú prostriedkami operačného systému. Ak vytvoríte vlákna nekontrolovateľne, môžu sa vám tieto prostriedky rýchlo minúť.

Kontextové prepínanie medzi vláknami vykonáva aj operačný systém - s cieľom napodobniť paralelizmus. Zjednodušené je to - čím viac vlákien sa objaví, tým menej času každé vlákno strávi skutočnou prácou.

Vzor Thread Pool pomáha šetriť zdroje vo viacvláknovej aplikácii a tiež obsahovať paralelizmus v určitých preddefinovaných limitoch.

Ak použijete fond vlákien, môžete napíšte svoj súbežný kód vo forme paralelných úloh a odošlite ich na vykonanie do inštancie fondu vlákien. Táto inštancia riadi niekoľko znovu použitých vlákien na vykonávanie týchto úloh.

Vzor vám to umožňuje kontrolovať počet vlákien, ktoré aplikácia vytvára, ich životného cyklu, ako aj na naplánovanie vykonania úloh a udržanie prichádzajúcich úloh v rade.

3. Fondy vlákien v Jave

3.1. Exekútori, Exekútor a ExecutorService

The Exekútori pomocná trieda obsahuje niekoľko metód na vytváranie predkonfigurovaných inštancií fondu vlákien pre vás. Tieto triedy sú dobrým miestom na začiatok - použite ich, ak nepotrebujete žiadne vlastné doladenie.

The Exekútor a ExecutorService rozhrania sa používajú na prácu s rôznymi implementáciami fondu vlákien v Jave. Spravidla by ste mali udržujte svoj kód odpojený od skutočnej implementácie fondu vlákien a používať tieto rozhrania v celej svojej aplikácii.

The Exekútor rozhranie má jeden vykonať spôsob predloženia Spustiteľné inštancie na vykonanie.

Tu je rýchly príklad ako môžete použiť Exekútori API na získanie Exekútor inštancia zálohovaná jednou oblasťou vlákien a neobmedzeným radom na postupné vykonávanie úloh. Tu vykonáme jednu úlohu, ktorá jednoducho vytlačí „Ahoj svet" na obrazovke. Úloha je odoslaná ako lambda (funkcia Java 8), o ktorej sa predpokladá Spustiteľné.

Exekútor exekútor = Executors.newSingleThreadExecutor (); executor.execute (() -> System.out.println ("Hello World"));

The ExecutorService rozhranie obsahuje veľké množstvo metód pre kontrola priebehu úloh a riadenie ukončenia služby. Pomocou tohto rozhrania môžete zadávať úlohy na vykonávanie a tiež kontrolovať ich vykonávanie pomocou vrátených Budúcnosť inštancia.

V nasledujúcom príklade, tvoríme ExecutorService, zadať úlohu a potom použiť vrátené Budúcnosť‘S dostať metóda čakať, kým sa zadaná úloha nedokončí a hodnota sa nevráti:

ExecutorService executorService = Executors.newFixedThreadPool (10); Budúca budúcnosť = executorService.submit (() -> „Hello World“); // niektoré operácie Výsledok reťazca = future.get ();

Samozrejme, v scenári z reálneho života sa vám obvykle nechce telefonovať future.get () hneď, ale odložte ho, až kým skutočne nebudete potrebovať hodnotu výpočtu.

The Predložiť metóda je preťažená Spustiteľné alebo Vyvolávateľná obe sú funkčné rozhrania a je možné ich odovzdať ako lambdas (počnúc jazykom Java 8).

SpustiteľnéJediná metóda nevyvolá výnimku a nevráti hodnotu. The Vyvolávateľná rozhranie môže byť pohodlnejšie, pretože nám umožňuje vyvolať výnimku a vrátiť hodnotu.

Nakoniec - nechať kompilátor odvodiť Vyvolávateľná typu, jednoducho vráťte hodnotu z lambda.

Ďalšie príklady použitia súboru ExecutorService rozhranie a futures, pozrite si „Sprievodcu Java ExecutorService“.

3.2. ThreadPoolExecutor

The ThreadPoolExecutor je rozšíriteľná implementácia fondu vlákien s množstvom parametrov a hákmi na jemné doladenie.

Hlavné konfiguračné parametre, o ktorých tu budeme diskutovať, sú: corePoolSize, maximumPoolSizea keepAliveTime.

Skupina pozostáva z pevného počtu základných vlákien, ktoré sú neustále uchovávané vo vnútri, a niektorých nadmerných vlákien, ktoré sa môžu vytvoriť a potom ukončiť, keď už nie sú potrebné. The corePoolSize parameter je počet vlákien jadra, ktoré budú inštancované a uchované v skupine. Keď príde nová úloha, ak sú všetky základné vlákna obsadené a interný front je plný, potom sa môže fond rozrásť až na maximumPoolSize.

The keepAliveTime Parameter je časový interval, po ktorý sú nadmerné vlákna (konkretizované nad corePoolSize) môžu existovať v pohotovostnom stave. V predvolenom nastavení je ThreadPoolExecutor na odstránenie berie do úvahy iba nepodložené vlákna. Na uplatnenie rovnakých pravidiel odstraňovania na základné vlákna môžeme použiť allowCoreThreadTimeOut (true) metóda.

Tieto parametre pokrývajú širokú škálu prípadov použitia, ale najtypickejšie konfigurácie sú preddefinované v Exekútori statické metódy.

Napríklad, newFixedThreadPool metóda vytvára a ThreadPoolExecutor s rovnakými corePoolSize a maximumPoolSize hodnoty parametrov a nula keepAliveTime. To znamená, že počet vlákien v tejto skupine vlákien je vždy rovnaký:

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool (2); executor.submit (() -> {Thread.sleep (1000); return null;}); executor.submit (() -> {Thread.sleep (1000); return null;}); executor.submit (() -> {Thread.sleep (1000); return null;}); assertEquals (2, executor.getPoolSize ()); assertEquals (1, executor.getQueue (). size ());

Vo vyššie uvedenom príklade vytvoríme inštanciu a ThreadPoolExecutor s pevným počtom vlákien 2. To znamená, že ak je počet súčasne spustených úloh vždy menší alebo rovný dvomi, okamžite sa vykonajú. Inak, niektoré z týchto úloh môžu byť zaradené do poradia, aby čakali na svoju príležitosť.

Vytvorili sme tri Vyvolávateľná úlohy, ktoré napodobňujú ťažkú ​​prácu tým, že spia 1 000 milisekúnd. Prvé dve úlohy sa vykonajú naraz a tretia bude musieť počkať v poradí. Môžeme to overiť zavolaním na getPoolSize () a getQueue (). size () metódy ihneď po zadaní úloh.

Ďalšia predkonfigurovaná ThreadPoolExecutor možno vytvoriť pomocou Executors.newCachedThreadPool () metóda. Táto metóda vôbec neprijíma určitý počet vlákien. The corePoolSize je v skutočnosti nastavená na 0 a maximumPoolSize je nastavený na Celé číslo.MAX_VALUE pre tento prípad. The keepAliveTime pre tento je 60 sekúnd.

Tieto hodnoty parametrov to znamenajú fond vlákien uložených v pamäti cache môže rásť bez hraníc, aby sa do nich zmestil ľubovoľný počet zadaných úloh. Ak však vlákna už nie sú potrebné, po 60 sekundách nečinnosti sa zlikvidujú. Typickým prípadom použitia je, keď máte vo svojej aplikácii veľa úloh, ktoré majú krátku životnosť.

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newCachedThreadPool (); executor.submit (() -> {Thread.sleep (1000); return null;}); executor.submit (() -> {Thread.sleep (1000); return null;}); executor.submit (() -> {Thread.sleep (1000); return null;}); assertEquals (3, executor.getPoolSize ()); assertEquals (0, executor.getQueue (). size ());

Veľkosť frontu v príklade vyššie bude vždy nulová, pretože interne a SynchronousQueue inštancia sa používa. V SynchronousQueue, páry vložiť a odstrániť operácie sa vyskytujú vždy súčasne, takže front vlastne nikdy nič neobsahuje.

The Executors.newSingleThreadExecutor () API vytvára ďalšiu typickú formu ThreadPoolExecutor obsahujúce jedno vlákno. Vykonávač s jedným vláknom je ideálny na vytvorenie slučky udalostí. The corePoolSize a maximumPoolSize parametre sa rovnajú 1 a keepAliveTime je nula.

Úlohy vo vyššie uvedenom príklade sa budú vykonávať postupne, takže hodnota príznaku bude 2 po dokončení úlohy:

Počítadlo AtomicInteger = nový AtomicInteger (); ExecutorService vykonávateľ = Executors.newSingleThreadExecutor (); Exekutor.submit (() -> {counter.set (1);}); Exekutor.submit (() -> {counter.compareAndSet (1, 2);});

Dodatočne toto ThreadPoolExecutor je zdobený nemenným obalom, takže ho po vytvorení nie je možné prekonfigurovať. Upozorňujeme, že aj to je dôvod, prečo ho nemôžeme použiť na a ThreadPoolExecutor.

3.3. ScheduledThreadPoolExecutor

The ScheduledThreadPoolExecutor rozširuje ThreadPoolExecutor triedy a tiež implementuje ScheduledExecutorService rozhranie s niekoľkými ďalšími metódami:

• harmonogram metóda umožňuje vykonať úlohu raz po zadanom oneskorení;
• scheduleAtFixedRate metóda umožňuje vykonať úlohu po zadanom počiatočnom oneskorení a potom ju vykonať opakovane s určitou dobou; the obdobie argumentom je čas merané medzi začiatočnými časmi úloh, takže miera vykonania je pevná;
• scheduleWithFixedDelay metóda je podobná ako scheduleAtFixedRate v tom, že opakovane vykoná danú úlohu, ale zadané oneskorenie je sa meria medzi koncom predchádzajúcej úlohy a začiatkom ďalšej; miera vykonania sa môže líšiť v závislosti od času potrebného na vykonanie danej úlohy.

The Executors.newScheduledThreadPool () metóda sa zvyčajne používa na vytvorenie a ScheduledThreadPoolExecutor s daným corePoolSize, bez obmedzenia maximumPoolSize a nula keepAliveTime. Tu je príklad, ako naplánovať vykonanie úlohy na 500 milisekúnd:

ScheduledExecutorService vykonávateľ = Executors.newScheduledThreadPool (5); executor.schedule (() -> {System.out.println ("Hello World");}, 500, TimeUnit.MILLISECONDS);

Nasledujúci kód ukazuje, ako vykonať úlohu po oneskorení 500 milisekúnd a potom ju opakovať každých 100 milisekúnd. Po naplánovaní úlohy počkáme, kým sa pomocou trikrát nespáli trikrát CountDownLatch zámok, potom ho zrušte pomocou Future.cancel () metóda.

Zámok CountDownLatch = nový CountDownLatch (3); ScheduledExecutorService vykonávateľ = Executors.newScheduledThreadPool (5); ScheduledFuture future = executor.scheduleAtFixedRate (() -> {System.out.println ("Hello World"); lock.countDown ();}, 500, 100, TimeUnit.MILLISECONDS); lock.await (1000, TimeUnit.MILLISECONDS); future.cancel (true);

3.4. ForkJoinPool

ForkJoinPool je centrálnou časťou vidlica / pripojiť sa framework zavedený v Jave 7. Rieši spoločný problém s plodiť viac úloh v rekurzívnych algoritmoch. Pomocou jednoduchého ThreadPoolExecutor, vlákna sa vám rýchlo minú, pretože každá úloha alebo podúloha vyžaduje na spustenie svoje vlastné vlákno.

V vidlica / pripojiť sa rámci sa môže objaviť akákoľvek úloha (vidlička) niekoľko podúloh a počkajte na ich dokončenie pomocou pripojiť sa metóda. Výhoda vidlica / pripojiť sa rámec je ten nevytvára nové vlákno pre každú úlohu alebo podúlohu, namiesto toho implementujúc algoritmus Work Stealing. Tento rámec je podrobne popísaný v článku „Sprievodca po rozhraní Fork / Join Framework v Jave“

Pozrime sa na jednoduchý príklad použitia ForkJoinPool prejsť stromom uzlov a vypočítať súčet všetkých listových hodnôt. Tu je jednoduchá implementácia stromu pozostávajúceho z uzla, int hodnota a množina podradených uzlov:

statická trieda TreeNode {int hodnota; Nastaviť deti; TreeNode (int hodnota, TreeNode ... deti) {this.value = hodnota; this.children = Sets.newHashSet (deti); }}

Teraz, ak chceme paralelne sčítať všetky hodnoty v strome, musíme implementovať a RecursiveTask rozhranie. Každá úloha dostane svoj vlastný uzol a svoju hodnotu pripočíta k súčtu hodnôt svojho deti. Na výpočet súčtu deti hodnoty, implementácia úlohy robí toto:

• potoky deti sada,
• mapy cez tento prúd a vytvárajú nové CountingTask pre každý prvok,
• vykoná každú čiastkovú úlohu rozvetvením,
• zhromažďuje výsledky volaním čísla pripojiť sa metóda pre každú vidlicovú úlohu,
• sčíta výsledky pomocou Collectors.summingInt zberateľ.

verejná statická trieda CountingTask rozširuje RecursiveTask {súkromný konečný uzol TreeNode; public CountingTask (uzol TreeNode) {this.node = uzol; } @Override chránený Integer compute () {návratový uzol.hodnota + uzol.children.stream () .map (childNode -> nový CountingTask (childNode) .fork ()) .collect (Collectors.summingInt (ForkJoinTask :: join)) ; }}

Kód na spustenie výpočtu na skutočnom strome je veľmi jednoduchý:

TreeNode strom = nový TreeNode (5, nový TreeNode (3), nový TreeNode (2, nový TreeNode (2), nový TreeNode (8)))); ForkJoinPool forkJoinPool = ForkJoinPool.commonPool (); int sum = forkJoinPool.invoke (nový CountingTask (strom));

4. Implementácia združenia Thread Pool v Guave

Guava je populárna knižnica nástrojov Google. Má veľa užitočných tried súbežnosti vrátane niekoľkých šikovných implementácií ExecutorService. Implementačné triedy nie sú prístupné pre priame inštancie alebo podtriedy, takže jediným vstupným bodom na vytváranie ich inštancií je Viac Exekútorov pomocná trieda.

4.1. Pridanie Guavy ako závislosti Maven

Pridajte do svojho súboru Maven pom nasledujúcu závislosť, aby ste do svojho projektu zahrnuli knižnicu Guava. Najnovšiu verziu knižnice Guava nájdete v úložisku Maven Central:

 com.google.guava guava 19.0 

4.2. Priamy exekútor a služba priameho exekútora

Niekedy chcete v závislosti na určitých podmienkach úlohu vykonať buď v aktuálnom vlákne, alebo vo fonde vlákien. Najradšej by ste použili jeden Exekútor rozhranie a stačí prepnúť implementáciu. Aj keď nie je také ťažké prísť s implementáciou Exekútor alebo ExecutorService ktorý vykonáva úlohy v aktuálnom vlákne, stále vyžaduje napísanie nejakého štandardného kódu.

Guava s radosťou poskytuje preddefinované inštancie pre nás.

Tu je príklad ktorá demonštruje vykonanie úlohy v rovnakom vlákne. Aj keď poskytnutá úloha spí 500 milisekúnd, je to tak blokuje aktuálne vláknoa výsledok je k dispozícii ihneď po vykonať hovor je ukončený:

Exekútor exekútor = MoreExecutors.directExecutor (); AtomicBoolean vykonaný = nový AtomicBoolean (); executor.execute (() -> {try {Thread.sleep (500);} catch (InterruptedException e) {e.printStackTrace ();} execut.set (true);}); assertTrue (execut.get ());

Inštancia vrátená directExecutor () metóda je vlastne statický singleton, takže použitie tejto metódy neposkytuje vôbec žiadne réžie na vytváranie objektov.

Mali by ste uprednostniť túto metódu pred MoreExecutors.newDirectExecutorService () pretože toto API vytvára pri každom hovore plnohodnotnú implementáciu služby vykonávateľa.

4.3. Ukončenie služieb exekútora

Ďalším častým problémom je vypnutie virtuálneho stroja zatiaľ čo fond vlákien stále vykonáva svoje úlohy. Ani pri zavedenom mechanizme zrušenia nie je zaručené, že sa úlohy budú správať pekne a zastavia svoju prácu, keď sa vypne služba vykonávateľa. To môže spôsobiť, že JVM bude visieť na neurčito, zatiaľ čo úlohy stále robia svoju prácu.

Na vyriešenie tohto problému spoločnosť Guava predstavuje rodinu vystupujúcich exekútorských služieb. Vychádzajú z vlákna démonov, ktoré končia spolu s JVM.

Tieto služby tiež pridávajú vypínací hák pomocou Runtime.getRuntime (). AddShutdownHook () metóda a zabrániť VM ukončiť sa na nakonfigurovaný čas pred tým, ako sa vzdá zavesených úloh.

V nasledujúcom príklade odosielame úlohu, ktorá obsahuje nekonečnú slučku, ale na ukončenie úlohy po ukončení VM používame službu exitujúceho exekútora s nakonfigurovaným časom 100 milisekúnd. Bez exitingExecutorService na mieste by táto úloha spôsobila, že VM bude visieť neurčito:

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool (5); ExecutorService executorService = MoreExecutors.getExitingExecutorService (exekútor, 100, TimeUnit.MILLISECONDS); executorService.submit (() -> {while (true) {}});

4.4. Počúvajúci dekoratéri

Počúvajúce dekoratéry vám umožňujú zabaliť ExecutorService a prijímať ListenableFuture inštancie po zadaní úlohy namiesto jednoduchých Budúcnosť inštancie. The ListenableFuture rozhranie sa rozširuje Budúcnosť a má jednu ďalšiu metódu addListener. Táto metóda umožňuje pridať poslucháča, ktorý sa volá pri budúcom dokončení.

Málokedy budete chcieť použiť ListenableFuture.addListener () metóda priamo, ale je nevyhnutné pre väčšinu pomocných metód v Budúcnosť úžitková trieda. Napríklad s Futures.allAsList () metóda môžete kombinovať niekoľko ListenableFuture inštancie v jednom ListenableFuture ktorý sa dokončí po úspešnom dokončení všetkých futures kombinovaných:

ExecutorService executorService = Executors.newCachedThreadPool (); ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator (executorService); ListenableFuture future1 = listeningExecutorService.submit (() -> „Hello“); ListenableFuture future2 = listeningExecutorService.submit (() -> "svet"); Reťazec pozdrav = Futures.allAsList (future1, future2) .get () .stream () .collect (Collectors.joining ("")); assertEquals ("Hello World", pozdrav);

5. Záver

V tomto článku sme sa zaoberali vzorom Thread Pool a jeho implementáciami v štandardnej knižnici Java a v knižnici Guava spoločnosti Google.

Zdrojový kód článku je k dispozícii na GitHub.