streamalism.org

1. Oversikt

I denne raske artikkelen ser vi på ConcurrentSkipListMap klasse fra java.util.concurrent pakke.

Denne konstruksjonen lar oss lage trådsikker logikk på en låsfri måte. Det er ideelt for problemer når vi ønsker å lage et uforanderlig øyeblikksbilde av dataene mens andre tråder fortsatt setter inn data på kartet.

Vi skal løse et problem med sortere en strøm av hendelser og få et øyeblikksbilde av hendelsene som ankom de siste 60 sekundene ved hjelp av konstruksjonen.

2. Strømmesorteringslogikk

La oss si at vi har en strøm av hendelser som kontinuerlig kommer fra flere tråder. Vi må kunne ta hendelser fra de siste 60 sekundene, og også hendelser som er eldre enn 60 sekunder.

La oss først definere strukturen til hendelsesdataene våre:

offentlig klasse Event {private ZonedDateTime eventTime; privat strenginnhold; // standardkonstruktører / getters}

Vi ønsker å holde arrangementene sortert ved hjelp av eventTime felt. For å oppnå dette ved hjelp av ConcurrentSkipListMap, vi må passere en Komparator til sin konstruktør mens den oppretter en forekomst av den:

ConcurrentSkipListMap events = new ConcurrentSkipListMap (Comparator.comparingLong (v -> v.toInstant (). ToEpochMilli ()));

Vi sammenligner alle ankomne hendelser ved hjelp av deres tidsstempler. Vi bruker comparingLong () metode og passere ekstraktfunksjonen som kan ta en lang tidsstempel fra ZonedDateTime.

Når arrangementene våre ankommer, trenger vi bare å legge dem til på kartet ved hjelp av sette() metode. Merk at denne metoden ikke krever noen eksplisitt synkronisering:

public void acceptEvent (Event event) {events.put (event.getEventTime (), event.getContent ()); }

De ConcurrentSkipListMap vil håndtere sorteringen av disse hendelsene under ved hjelp av Komparator som ble overført til den i konstruktøren.

De mest bemerkelsesverdige proffene til ConcurrentSkipListMap er metodene som kan lage et uforanderlig øyeblikksbilde av dataene på en låsfri måte. For å få alle hendelser som ankom i løpet av det siste øyeblikket, kan vi bruke tailMap () metode og passere tiden vi ønsker å få elementer fra:

public ConcurrentNavigableMap getEventsFromLastMinute () {return events.tailMap (ZonedDateTime.now (). minusMinutes (1)); } 

Den vil returnere alle hendelser fra siste øyeblikk. Det vil være et uforanderlig øyeblikksbilde, og det som er viktigst er at andre skrivetråder kan legge til nye hendelser i ConcurrentSkipListMap uten behov for å gjøre eksplisitt låsing.

Vi kan nå få alle hendelser som ankom senere ett minutt fra nå - ved å bruke headMap () metode:

public ConcurrentNavigableMap getEventsOlderThatOneMinute () {return events.headMap (ZonedDateTime.now (). minusMinutes (1)); }

Dette vil gi et uforanderlig øyeblikksbilde av alle hendelser som er eldre enn ett minutt. Alle ovennevnte metoder tilhører EventWindowSort klasse, som vi bruker i neste avsnitt.

3. Testing av sorteringsstrømlogikken

Når vi implementerte vår sorteringslogikk ved hjelp av ConcurrentSkipListMap, det kan vi nå test det ved å lage to forfattertråder som vil sende hundre arrangementer hver:

ExecutorService executorService = Executors.newFixedThreadPool (3); EventWindowSort eventWindowSort = nytt EventWindowSort (); int numberOfThreads = 2; Kjørbar produsent = () -> IntStream .rangeClosed (0, 100) .forEach (index -> eventWindowSort.acceptEvent (new Event (ZonedDateTime.now (). MinusSeconds (index), UUID.randomUUID (). ToString ())) ); for (int i = 0; i <numberOfThreads; i ++) {executorService.execute (produsent); } 

Hver tråd påkaller acceptEvent () metode, sender hendelsene som har eventTime fra nå til “nå minus hundre sekunder”.

I mellomtiden kan vi påberope oss getEventsFromLastMinute () metode som vil returnere øyeblikksbildet av hendelser som er innenfor ett minutt-vinduet:

ConcurrentNavigableMap eventsFromLastMinute = eventWindowSort.getEventsFromLastMinute ();

Antall hendelser i eventsFromLastMinute vil variere i hver testkjøring avhengig av hastigheten som produsenttrådene skal sende hendelsene til EventWindowSort. Vi kan hevde at det ikke er en eneste hendelse i det returnerte øyeblikksbildet som er eldre enn ett minutt:

lange hendelserOlderThanOneMinute = eventsFromLastMinute .entrySet () .stream () .filter (e -> e.getKey (). isBefore (ZonedDateTime.now (). minusMinutes (1))) .count (); assertEquals (eventsOlderThanOneMinute, 0);

Og at det er mer enn null hendelser i øyeblikksbildet som ligger innenfor ett minutt-vinduet:

long eventYoungerThanOneMinute = eventsFromLastMinute .entrySet () .stream () .filter (e -> e.getKey (). isAfter (ZonedDateTime.now (). minusMinutes (1))) .count (); assertTrue (eventYoungerThanOneMinute> 0);

Våre getEventsFromLastMinute () bruker tailMap () under.

La oss teste nå getEventsOlderThatOneMinute () som bruker headMap () metoden fra ConcurrentSkipListMap:

ConcurrentNavigableMap eventsFromLastMinute = eventWindowSort.getEventsOlderThatOneMinute ();

Denne gangen får vi et øyeblikksbilde av hendelser som er eldre enn ett minutt. Vi kan hevde at det er mer enn null av slike hendelser:

lange hendelserOlderThanOneMinute = eventsFromLastMinute .entrySet () .stream () .filter (e -> e.getKey (). isBefore (ZonedDateTime.now (). minusMinutes (1))) .count (); assertTrue (eventsOlderThanOneMinute> 0);

Og neste, at det ikke er en eneste begivenhet som er fra siste øyeblikk:

long eventYoungerThanOneMinute = eventsFromLastMinute .entrySet () .stream () .filter (e -> e.getKey (). isAfter (ZonedDateTime.now (). minusMinutes (1))) .count (); assertEquals (eventYoungerThanOneMinute, 0);

Det viktigste å merke seg er at vi kan ta øyeblikksbildet av data mens andre tråder fortsatt legger til nye verdier til ConcurrentSkipListMap.

4. Konklusjon

I denne raske opplæringen så vi på det grunnleggende om ConcurrentSkipListMap, sammen med noen praktiske eksempler.

Vi utnyttet den høye ytelsen til ConcurrentSkipListMap å implementere en ikke-blokkerende algoritme som kan tjene oss et uforanderlig øyeblikksbilde av data selv om det samtidig er flere tråder som oppdaterer kartet.

Implementeringen av alle disse eksemplene og kodebiter finner du i GitHub-prosjektet; dette er et Maven-prosjekt, så det skal være enkelt å importere og kjøre som det er.