streamalism.org

1. Oversigt

I denne artikel ser vi på to konstruktioner fra java.util.concurrent pakke: LongAdder og Lang akkumulator.

Begge er skabt til at være meget effektive i miljøet med flere tråde, og begge udnytter meget kloge taktikker låsefri og stadig trådfri.

2. LongAdder

Lad os overveje nogle logik, der forøger nogle værdier meget ofte, hvor vi bruger en AtomicLong kan være en flaskehals. Dette bruger en sammenlignings-og-swap-operation, som - under stærk strid - kan føre til mange spildte CPU-cyklusser.

LongAdderbruger derimod et meget smart trick til at reducere strid mellem tråde, når disse øges.

Når vi vil øge en forekomst af LongAdder, vi er nødt til at kalde stigning () metode. Denne implementering holder en række tællere, der kan vokse efter behov.

Og så når flere tråde ringer stigning (), vil arrayet være længere. Hver post i arrayet kan opdateres separat - hvilket reducerer påstanden. På grund af dette faktum, LongAdder er en meget effektiv måde at forøge en tæller fra flere tråde.

Lad os oprette en forekomst af LongAdder klasse og opdatere den fra flere tråde:

LongAdder-tæller = ny LongAdder (); ExecutorService executorService = Executors.newFixedThreadPool (8); int numberOfThreads = 4; int numberOfIncrements = 100; Kørbar incrementAction = () -> IntStream .range (0, numberOfIncrements) .forEach (i -> counter.increment ()); for (int i = 0; i <numberOfThreads; i ++) {executorService.execute (incrementAction); }

Resultatet af tælleren i LongAdder er ikke tilgængelig, før vi ringer til sum() metode. Denne metode gentager alle værdier i nedenstående array og summerer disse værdier, der returnerer den korrekte værdi. Vi skal dog være forsigtige, fordi opkaldet til sum() metoden kan være meget dyr:

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

Nogle gange, når vi ringer sum(), vi ønsker at rydde al tilstand, der er forbundet med forekomsten af LongAdder og start med at tælle fra starten. Vi kan bruge sumThenReset () metode til at opnå det:

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

Bemærk, at det efterfølgende opkald til sum() metode returnerer nul, hvilket betyder, at staten blev nulstillet.

Desuden giver Java også DoubleAdder at opretholde en summering af dobbelt værdier med en lignende API til LongAdder.

3. LongAkkumulator

Lang akkumulator er også en meget interessant klasse - som giver os mulighed for at implementere en låsefri algoritme i en række scenarier. For eksempel kan den bruges til at akkumulere resultater i henhold til den leverede LongBinaryOperator - dette fungerer på samme måde som reducere() drift fra Stream API.

Forekomsten af Lang akkumulator kan oprettes ved at levere LongBinaryOperator og den oprindelige værdi for dens konstruktør. Den vigtige ting at huske det Lang akkumulator fungerer korrekt, hvis vi forsyner det med en kommutativ funktion, hvor rækkefølgen af ​​akkumulering ikke betyder noget.

LongAccumulator akkumulator = ny LongAccumulator (Long :: sum, 0L);

Vi opretter en Lang akkumulator which tilføjer en ny værdi til den værdi, der allerede var i akkumulatoren. Vi indstiller startværdien af Lang akkumulator til nul, så i det første opkald af ophobe() metode, den previousValue har en nulværdi.

Lad os påberåbe sig ophobe() metode fra flere tråde:

int numberOfThreads = 4; int numberOfIncrements = 100; Kørbar accumulateAction = () -> IntStream .rangeClosed (0, numberOfIncrements) .forEach (akkumulator :: akkumulere); for (int i = 0; i <numberOfThreads; i ++) {executorService.execute (accumulateAction); }

Læg mærke til, hvordan vi sender et tal som et argument til ophobe() metode. Denne metode vil påberåbe sig vores sum() fungere.

Det Lang akkumulator bruger sammenligning-og-swap-implementeringen - hvilket fører til disse interessante semantik.

For det første udfører den en handling defineret som en LongBinaryOperator, og derefter kontrollerer det, om previousValue ændret. Hvis det blev ændret, udføres handlingen igen med den nye værdi. Hvis ikke, lykkes det at ændre den værdi, der er gemt i akkumulatoren.

Vi kan nu hævde, at summen af ​​alle værdier fra alle iterationer var 20200:

assertEquals (accumulator.get (), 20200);

Interessant nok giver Java også DoubleAccumulator med samme formål og API men til dobbelt værdier.

4. Dynamisk stripning

Alle implementeringer af addere og akkumulatorer i Java arver fra en interessant baseklasse kaldet Stribet64. I stedet for kun at bruge en værdi til at opretholde den aktuelle tilstand bruger denne klasse en række stater til at distribuere påstanden til forskellige hukommelsesplaceringer.

Her er en simpel skildring af hvad Stribet64 gør:

Forskellige tråde opdaterer forskellige hukommelsesplaceringer. Da vi bruger en matrix (dvs. striber) af tilstande, kaldes denne idé dynamisk stribe. Interessant nok Stribet64 er opkaldt efter denne idé og det faktum, at den fungerer på 64-bit datatyper.

Vi forventer, at dynamisk striping forbedrer den samlede ydeevne. Imidlertid kan den måde, som JVM tildeler disse stater på, have en kontraproduktiv effekt.

For at være mere specifik kan JVM allokere disse stater nær hinanden i bunken. Dette betyder, at nogle få stater kan opholde sig i den samme CPU-cache-linje. Derfor, opdatering af en hukommelsesplacering kan medføre, at en cache går glip af dens nærliggende stater. Dette fænomen, kendt som falsk deling, vil skade forestillingen.

For at forhindre falsk deling. det Stribet64 implementering tilføjer tilstrækkelig polstring omkring hver stat for at sikre, at hver stat befinder sig i sin egen cache-linje:

Det @ Contended annotation er ansvarlig for at tilføje denne polstring. Polstringen forbedrer ydeevnen på bekostning af mere hukommelsesforbrug.

5. Konklusion

I denne hurtige vejledning kiggede vi på LongAdder og Lang akkumulator og vi har vist, hvordan man bruger begge konstruktioner til at implementere meget effektive og låsefrie løsninger.

Implementeringen af ​​alle disse eksempler og kodestykker findes i GitHub-projektet - dette er et Maven-projekt, så det skal være let at importere og køre, som det er.