streamalism.org

1. Oversigt

I denne hurtige artikel ser vi på ConcurrentSkipListMap klasse fra java.util.concurrent pakke.

Denne konstruktion giver os mulighed for at skabe trådsikker logik på en låsefri måde. Det er ideelt til problemer, når vi ønsker at lave et uforanderligt øjebliksbillede af dataene, mens andre tråde stadig indsætter data på kortet.

Vi vil løse et problem med sortering af en strøm af begivenheder og få et øjebliksbillede af de begivenheder, der ankom i de sidste 60 sekunder ved hjælp af denne konstruktion.

2. Streamsorteringslogik

Lad os sige, at vi har en strøm af begivenheder, der løbende kommer fra flere tråde. Vi skal være i stand til at tage begivenheder fra de sidste 60 sekunder og også begivenheder, der er ældre end 60 sekunder.

Lad os først definere strukturen på vores begivenhedsdata:

offentlig klasse begivenhed {privat ZonedDateTime begivenhedstid; privat strengindhold; // standard konstruktører / getters}

Vi ønsker at holde vores begivenheder sorteret ved hjælp af eventTime Mark. For at opnå dette ved hjælp af ConcurrentSkipListMap, vi har brug for at videregive en Komparator til sin konstruktør, mens den opretter en forekomst af den:

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

Vi sammenligner alle ankomne begivenheder ved hjælp af deres tidsstempler. Vi bruger comparingLong () metode og videregive ekstraktfunktionen, der kan tage en lang tidsstempel fra ZonedDateTime.

Når vores begivenheder ankommer, behøver vi kun tilføje dem til kortet ved hjælp af sætte() metode. Bemærk, at denne metode ikke kræver nogen eksplicit synkronisering:

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

Det ConcurrentSkipListMap håndterer sorteringen af ​​disse begivenheder nedenunder ved hjælp af Komparator der blev sendt til det i konstruktøren.

De mest bemærkelsesværdige fordele ved ConcurrentSkipListMap er de metoder, der kan lave et uforanderligt øjebliksbillede af dets data på en låsefri måde. For at få alle begivenheder, der ankom inden for det sidste minut, kan vi bruge tailMap () metode og passere den tid, hvorfra vi ønsker at få elementer:

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

Det returnerer alle begivenheder fra det sidste minut. Det vil være et uforanderligt øjebliksbillede, og hvad der er det vigtigste er, at andre skrivetråde kan tilføje nye begivenheder til ConcurrentSkipListMap uden behov for eksplicit låsning.

Vi kan nu få alle begivenheder, der ankom senere et minut fra nu - ved hjælp af headMap () metode:

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

Dette returnerer et uforanderligt øjebliksbillede af alle begivenheder, der er ældre end et minut. Alle ovenstående metoder hører til EventWindowSort klasse, som vi bruger i næste afsnit.

3. Test af sorteringsstrømslogikken

Når vi implementerede vores sorteringslogik ved hjælp af ConcurrentSkipListMap, det kan vi nu test det ved at oprette to forfattertråde der vil sende hundrede begivenheder hver:

ExecutorService executorService = Executors.newFixedThreadPool (3); EventWindowSort eventWindowSort = ny EventWindowSort (); int numberOfThreads = 2; Kører producent = () -> 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 (producer); } 

Hver tråd påberåber sig acceptEvent () metode, sender de begivenheder, der har eventTime fra nu til “nu minus hundrede sekunder”.

I mellemtiden kan vi påberåbe os getEventsFromLastMinute () metode, der returnerer øjebliksbillede af begivenheder, der er inden for et minuts vindue:

ConcurrentNavigableMap eventsFromLastMinute = eventWindowSort.getEventsFromLastMinute ();

Antallet af begivenheder i eventsFromLastMinute vil variere i hver testkørsel afhængigt af den hastighed, hvormed producenttrådene sender begivenhederne til EventWindowSort. Vi kan hævde, at der ikke er en enkelt begivenhed i det returnerede øjebliksbillede, der er ældre end et minut:

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

Og at der er mere end nul begivenheder i snapshotet, der ligger inden for et minuts vindue:

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

Vores getEventsFromLastMinute () bruger tailMap () under.

Lad os teste nu getEventsOlderThatOneMinute () der bruger headMap () metode fra ConcurrentSkipListMap:

ConcurrentNavigableMap eventsFromLastMinute = eventWindowSort.getEventsOlderThatOneMinute ();

Denne gang får vi et øjebliksbillede af begivenheder, der er ældre end et minut. Vi kan hævde, at der er mere end nul af sådanne begivenheder:

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

Og dernæst, at der ikke er en enkelt begivenhed, der er inden for sidste øjeblik:

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

Det vigtigste at bemærke er, at vi kan tage øjebliksbillede af data, mens andre tråde stadig tilføjer nye værdier til ConcurrentSkipListMap.

4. Konklusion

I denne hurtige vejledning kiggede vi på det grundlæggende i ConcurrentSkipListMapsammen med nogle praktiske eksempler.

Vi udnyttede den høje ydeevne for ConcurrentSkipListMap at implementere en ikke-blokerende algoritme, der kan tjene os til et uforanderligt øjebliksbillede af data, selvom flere tråde samtidig opdaterer kortet.

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.