streamalism.org

1. Overzicht

In dit korte artikel zullen we kijken naar de ConcurrentSkipListMap klasse uit de java.util.concurrent pakket.

Met deze constructie kunnen we threadveilige logica op een vergrendelingsvrije manier creëren. Het is ideaal voor problemen wanneer we een onveranderlijke momentopname van de gegevens willen maken terwijl andere threads nog steeds gegevens in de kaart invoegen.

We zullen een probleem oplossen van het sorteren van een stroom gebeurtenissen en het verkrijgen van een momentopname van de gebeurtenissen die in de afgelopen 60 seconden zijn aangekomen met behulp van dat construct.

2. Logica voor het sorteren van streams

Laten we zeggen dat we een stroom van evenementen hebben die voortdurend uit meerdere threads komen. We moeten gebeurtenissen van de afgelopen 60 seconden kunnen opnemen, en ook gebeurtenissen die ouder zijn dan 60 seconden.

Laten we eerst de structuur van onze evenementgegevens definiëren:

openbare klasse Event {privé ZonedDateTime eventTime; privé String-inhoud; // standard constructors / getters}

We willen onze evenementen gesorteerd houden met behulp van de eventTime veld. Om dit te bereiken met behulp van de ConcurrentSkipListMap, we moeten een Comparator naar zijn constructor tijdens het maken van een instantie ervan:

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

We vergelijken alle aangekomen evenementen met behulp van hun tijdstempels. We gebruiken de vergelijkenLong () methode en het doorgeven van de extractiefunctie die een lang tijdstempel van het ZonedDateTime.

Wanneer onze evenementen arriveren, hoeven we ze alleen aan de kaart toe te voegen met de leggen() methode. Merk op dat deze methode geen expliciete synchronisatie vereist:

openbare ongeldige acceptEvent (gebeurtenis gebeurtenis) {events.put (event.getEventTime (), event.getContent ()); }

De ConcurrentSkipListMap zal het sorteren van die gebeurtenissen eronder afhandelen met behulp van de Comparator dat eraan is doorgegeven in de constructor.

De meest opvallende voordelen van de ConcurrentSkipListMap zijn de methoden die een onveranderlijke momentopname van de gegevens op een vergrendelingsvrije manier kunnen maken. Om alle evenementen te krijgen die in de afgelopen minuut zijn aangekomen, kunnen we de tailMap () methode en geef de tijd door waaruit we elementen willen halen:

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

Alle gebeurtenissen van de afgelopen minuut worden geretourneerd. Het zal een onveranderlijke momentopname zijn en het belangrijkste is dat andere schrijfthreads nieuwe gebeurtenissen kunnen toevoegen aan het ConcurrentSkipListMap zonder enige noodzaak om expliciet te vergrendelen.

We kunnen nu alle gebeurtenissen ophalen die later binnen een minuut zijn aangekomen - door de headMap () methode:

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

Hiermee wordt een onveranderlijke momentopname geretourneerd van alle gebeurtenissen die ouder zijn dan één minuut. Alle bovenstaande methoden behoren tot de EventWindowSort class, die we in de volgende sectie zullen gebruiken.

3. Testen van de sorteerstroomlogica

Nadat we onze sorteerlogica hadden geïmplementeerd met behulp van de ConcurrentSkipListMap, we kunnen nu test het door twee schrijfthreads te maken die elk honderd evenementen sturen:

ExecutorService executorService = Executors.newFixedThreadPool (3); EventWindowSort eventWindowSort = nieuwe EventWindowSort (); int numberOfThreads = 2; Runnable producer = () -> IntStream .rangeClosed (0, 100) .forEach (index -> eventWindowSort.acceptEvent (nieuwe gebeurtenis (ZonedDateTime.now (). MinusSeconds (index), UUID.randomUUID (). ToString ())) ); voor (int i = 0; i <numberOfThreads; i ++) {executorService.execute (producer); } 

Elke thread roept de acceptEvent () methode, het verzenden van de gebeurtenissen die eventTime van nu naar "nu min honderd seconden".

In de tussentijd kunnen we een beroep doen op het getEventsFromLastMinute () methode die de momentopname van gebeurtenissen binnen het venster van één minuut retourneert:

ConcurrentNavigableMap eventsFromLastMinute = eventWindowSort.getEventsFromLastMinute ();

Het aantal gebeurtenissen in het eventsFromLastMinute varieert in elke testrun, afhankelijk van de snelheid waarmee de producer-threads de gebeurtenissen naar de EventWindowSort. We kunnen stellen dat er geen enkele gebeurtenis in de geretourneerde snapshot is die ouder is dan één minuut:

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

En dat er meer dan nul gebeurtenissen in de momentopname zijn die binnen het venster van één minuut vallen:

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

Onze getEventsFromLastMinute () gebruikt de tailMap () onder.

Laten we nu het getEventsOlderThatOneMinute () dat is met behulp van de headMap () methode van de ConcurrentSkipListMap:

ConcurrentNavigableMap eventsFromLastMinute = eventWindowSort.getEventsOlderThatOneMinute ();

Deze keer krijgen we een momentopname van gebeurtenissen die ouder zijn dan een minuut. We kunnen stellen dat er meer dan nul van dergelijke gebeurtenissen zijn:

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

En vervolgens, dat er geen enkele gebeurtenis is die van de laatste minuut is:

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

Het belangrijkste om op te merken is dat we kunnen de momentopname van gegevens maken terwijl andere threads nog steeds nieuwe waarden toevoegen naar de ConcurrentSkipListMap.

4. Conclusie

In deze korte tutorial hebben we de basisprincipes van het ConcurrentSkipListMap, samen met enkele praktische voorbeelden.

We hebben gebruik gemaakt van de hoge prestaties van de ConcurrentSkipListMap om een ​​niet-blokkerend algoritme te implementeren dat ons een onveranderlijke momentopname van gegevens kan bieden, zelfs als tegelijkertijd meerdere threads de kaart bijwerken.

De implementatie van al deze voorbeelden en codefragmenten is te vinden in het GitHub-project; dit is een Maven-project, dus het moet gemakkelijk te importeren en uit te voeren zijn zoals het is.