streamalism.org

1. Prehľad

V tomto článku sa pozrieme na dva konštrukty z java.util.concurrent balenie: LongAdder a LongAccumulator.

Oba sú vytvorené tak, aby boli veľmi efektívne v prostredí viacerých vlákien, a obidve využívajú veľmi šikovnú taktiku bez zámku a stále zostávajú bezpečné pre vlákna.

2. LongAdder

Uvažujme o logike, ktorá veľmi často zvyšuje niektoré hodnoty, kde použitie znaku AtomicLong môže byť prekážkou. Používa sa operácia porovnania a výmeny, ktorá - pod silným sporom - môže viesť k mnohým zbytočným cyklom CPU.

LongAdder, na druhej strane, používa veľmi šikovný trik na zníženie sporu medzi vláknami, keď ich zvyšujú.

Keď chceme zvýšiť inštanciu LongAdder, musíme zavolať prírastok () metóda. Táto implementácia udržuje rad počítadiel, ktoré môžu rásť na požiadanie.

A tak, keď volá viac vlákien prírastok (), pole bude dlhšie. Každý záznam v poli je možné aktualizovať osobitne - znižuje sa tak tvrdenie. Z tohto dôvodu LongAdder je veľmi efektívny spôsob zvýšenia počítadla z viacerých vlákien.

Vytvorme inštanciu súboru LongAdder triedy a aktualizovať ju z viacerých vlákien:

Počítadlo LongAdder = nový LongAdder (); ExecutorService executorService = Executors.newFixedThreadPool (8); int numberOfThreads = 4; int numberOfIncrements = 100; Spustiteľná incrementAction = () -> IntStream. Range (0, numberOfIncrements) .forEach (i -> counter.increment ()); pre (int i = 0; i <numberOfThreads; i ++) {executorService.execute (incrementAction); }

Výsledok počítadla v LongAdder nie je k dispozícii, kým nezavoláme suma () metóda. Táto metóda bude iterovať nad všetkými hodnotami poľa pod a spočíta tie hodnoty, ktoré vrátia správnu hodnotu. Musíme byť však opatrní, pretože výzva na suma () metóda môže byť veľmi nákladná:

assertEquals (counter.sum (), numberOfIncrements * numberOfThreads);

Niekedy potom, ako zavoláme suma (), chceme vyčistiť všetok stav, ktorý je spojený s inštanciou LongAdder a začni počítať od začiatku. Môžeme použiť sumThenReset () spôsob, ako to dosiahnuť:

assertEquals (counter.sumThenReset (), numberOfIncrements * numberOfThreads); assertEquals (counter.sum (), 0);

Upozorňujeme, že následný hovor na číslo suma () metóda vracia nulu, čo znamená, že stav bol úspešne vynulovaný.

Okrem toho poskytuje aj Java DoubleAdder zachovať súčet dvojitý hodnoty s podobným API ako LongAdder.

3. LongAccumulator

LongAccumulator je tiež veľmi zaujímavá trieda - ktorá nám umožňuje implementovať algoritmus bez blokovania v mnohých scenároch. Môže sa napríklad použiť na zhromažďovanie výsledkov podľa dodaného LongBinaryOperator - toto funguje podobne ako znížiť () operácia z Stream API.

Inštancia LongAccumulator je možné vytvoriť dodaním LongBinaryOperator a počiatočná hodnota pre jeho konštruktor. Je dôležité si to uvedomiť LongAccumulator bude fungovať správne, ak jej dodáme komutatívnu funkciu, na ktorej nezáleží na poradí akumulácie.

Akumulátor LongAccumulator = nový LongAccumulator (Long :: sum, 0L);

Vytvárame LongAccumulator which pridá novú hodnotu k hodnote, ktorá už bola v akumulátore. Nastavujeme počiatočnú hodnotu LongAccumulator na nulu, takže pri prvom volaní čísla akumulovať () metóda, previousValue bude mať nulovú hodnotu.

Vyvolame akumulovať () metóda z viacerých vlákien:

int numberOfThreads = 4; int numberOfIncrements = 100; Spustiteľná akumuláciaAction = () -> IntStream .rangeClosed (0, numberOfIncrements) .forEach (akumulátor :: akumulovať); pre (int i = 0; i <numberOfThreads; i ++) {vykonávateľService.execute (akumulovaťAkcia); }

Všimnite si, ako odovzdávame číslo ako argument akumulovať () metóda. Táto metóda sa bude odvolávať na našu suma () funkcie.

The LongAccumulator používa implementáciu porovnania a výmeny (swap), ktorá vedie k tejto zaujímavej sémantike.

Najskôr vykoná akciu definovanú ako a LongBinaryOperator, a potom skontroluje, či previousValue zmenil. Ak bola zmenená, akcia sa vykoná znova s ​​novou hodnotou. Ak nie, uspeje v zmene hodnoty, ktorá je uložená v akumulátore.

Teraz môžeme tvrdiť, že súčet všetkých hodnôt zo všetkých iterácií bol 20200:

assertEquals (acculator.get (), 20200);

Je zaujímavé, že aj Java DoubleAccumulator s rovnakým účelom a API, ale pre dvojitý hodnoty.

4. Dynamické prúžkovanie

Všetky implementácie sčítačky a akumulátora v Jave dedia zo zaujímavej základnej triedy s názvom 64. Namiesto použitia iba jednej hodnoty na udržanie aktuálneho stavu používa táto trieda pole stavov na distribúciu sporu do rôznych pamäťových miest.

Tu je jednoduché znázornenie toho, čo 64 robí:

Rôzne vlákna aktualizujú rôzne miesta v pamäti. Pretože používame pole (tj. Pruhy) stavov, táto myšlienka sa nazýva dynamické pruhovanie. Je zaujímavé, že 64 je pomenovaný po tejto myšlienke a skutočnosti, že pracuje na 64-bitových dátových typoch.

Očakávame, že dynamické pruhy zlepšia celkový výkon. Spôsob, akým JVM prideľuje tieto štáty, však môže mať kontraproduktívny účinok.

Presnejšie povedané, JVM môže v halde prideliť tieto štáty blízko seba. To znamená, že niekoľko štátov sa môže nachádzať v rovnakom riadku medzipamäte CPU. Preto aktualizácia jedného miesta v pamäti môže spôsobiť vynechanie medzipamäte v jej blízkych stavoch. Tento jav, známy ako nepravé zdieľanie, poškodí výkon.

Aby sa zabránilo falošnému zdieľaniu. the 64 implementácia pridáva dostatok polstrovania okolo každého stavu, aby sa zabezpečilo, že každý štát sa nachádza vo svojom vlastnom riadku vyrovnávacej pamäte:

The @ Contended za pridanie tejto výplne je zodpovedná anotácia. Výplň zvyšuje výkon na úkor väčšej spotreby pamäte.

5. Záver

V tomto rýchlom návode sme sa pozreli na LongAdder a LongAccumulator a ukázali sme, ako použiť obidva konštrukty na implementáciu veľmi efektívnych a bezblokových riešení.

Implementáciu všetkých týchto príkladov a útržkov kódu nájdete v projekte GitHub - jedná sa o projekt Maven, takže by malo byť ľahké ho importovať a spustiť tak, ako je.