streamalism.org

1. Pregled

U ovom ćemo kratkom članku pogledati ConcurrentSkipListMap razred iz java.util.concurrent paket.

Ova nam konstrukcija omogućuje stvaranje logike sigurne u nitima na način bez zaključavanja. Idealno je za probleme kada želimo napraviti nepromjenjivu snimku podataka dok druge niti još uvijek ubacuju podatke na kartu.

Riješit ćemo problem sortiranje toka događaja i dobivanje snimke događaja koji su stigli u zadnjih 60 sekundi pomoću te konstrukcije.

2. Logika sortiranja toka

Recimo da imamo tok događaja koji neprestano dolaze iz više niti. Moramo biti u mogućnosti uzimati događaje u zadnjih 60 sekundi, kao i događaje starije od 60 sekundi.

Prvo definirajmo strukturu podataka o našem događaju:

javna klasa Event {private ZonedDateTime eventTime; privatni sadržaj niza; // standardni konstruktori / getteri}

Želimo da naši događaji budu sortirani pomoću eventTime polje. Da biste to postigli pomoću ConcurrentSkipListMap, trebamo proći a Usporednik njegovom konstruktoru dok stvara njegovu instancu:

ConcurrentSkipListMap događaji = novi ConcurrentSkipListMap (Comparator.comparingLong (v -> v.toInstant (). ToEpochMilli ()));

Usporedit ćemo sve pristigle događaje koristeći njihove vremenske oznake. Koristimo uspoređujućiLong () metoda i prosljeđivanje funkcije ekstrakcije koja može potrajati a dugo vremenska oznaka iz ZonedDateTime.

Kada naši događaji stignu, trebamo ih samo dodati na kartu pomoću staviti() metoda. Napominjemo da ova metoda ne zahtijeva izričitu sinkronizaciju:

javna praznina acceptEvent (događaj događaja) {events.put (event.getEventTime (), event.getContent ()); }

The ConcurrentSkipListMap će se nositi sa sortiranjem tih događaja ispod pomoću Usporednik koja mu je proslijeđena u konstruktoru.

Najistaknutiji profesionalci ConcurrentSkipListMap su metode koje mogu napraviti nepromjenjivu snimku svojih podataka na način bez zaključavanja. Da bismo dobili sve događaje koji su stigli u prošloj minuti, možemo koristiti tailMap () metodu i prođe vrijeme iz kojeg želimo dobiti elemente:

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

Vratit će sve događaje iz prošle minute. Bit će to nepromjenjiva snimka i najvažnije je da druge niti za pisanje mogu dodavati nove događaje u ConcurrentSkipListMap bez ikakve potrebe za eksplicitnim zaključavanjem.

Sada možemo dobiti sve događaje koji su stigli kasnije, nakon jedne minute - pomoću headMap () metoda:

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

To će vratiti nepromjenjivu snimku svih događaja starijih od jedne minute. Sve gore navedene metode pripadaju EventWindowSort razred, koji ćemo upotrijebiti u sljedećem odjeljku.

3. Testiranje logike toka sortiranja

Jednom kada smo implementirali našu logiku sortiranja pomoću ConcurrentSkipListMap, sada možemo testirajte ga stvaranjem dviju niti pisanja koji će poslati po sto događaja:

ExecutorService executorService = Izvršitelji.newFixedThreadPool (3); EventWindowSort eventWindowSort = novo EventWindowSort (); int numberOfThreads = 2; Izvodljivi proizvođač = () -> IntStream .rangeClosed (0, 100) .forEach (index -> eventWindowSort.acceptEvent (novi događaj (ZonedDateTime.now (). MinusSeconds (indeks), UUID.randomUUID (). ToString ())) ); for (int i = 0; i <numberOfThreads; i ++) {executorService.execute (producent); } 

Svaka nit poziva na acceptEvent () metodom, slanjem događaja koji jesu eventTime od sada do "sada minus sto sekundi".

U međuvremenu se možemo pozvati na getEventsFromLastMinute () metoda koja će vratiti snimak događaja koji su unutar jednogminutnog prozora:

ConcurrentNavigableMap eventsFromLastMinute = eventWindowSort.getEventsFromLastMinute ();

Broj događaja u eventsFromLastMinute varirat će u svakom probnom radu, ovisno o brzini kojom će proizvođačke niti slati događaje na EventWindowSort. Možemo tvrditi da u vraćenom snimku nema niti jednog događaja koji je stariji od jedne minute:

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

I da na snimci ima više od nule događaja koji se nalaze u roku od jedne minute:

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

Naše getEventsFromLastMinute () koristi tailMap () ispod.

Isprobajmo sada getEventsOlderThatOneMinute () koji koristi headMap () metoda iz ConcurrentSkipListMap:

ConcurrentNavigableMap eventsFromLastMinute = eventWindowSort.getEventsOlderThatOneMinute ();

Ovaj put dobivamo snimku događaja starijih od jedne minute. Možemo ustvrditi da takvih događaja ima više od nule:

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

I sljedeće, da nema niti jednog događaja koji nije iz zadnje minute:

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

Najvažnije je napomenuti da je možemo napraviti snimku podataka dok druge niti još dodaju nove vrijednosti prema ConcurrentSkipListMap.

4. Zaključak

U ovom smo brzom vodiču pogledali osnove ConcurrentSkipListMap, zajedno s nekoliko praktičnih primjera.

Potaknuli smo visoke performanse ConcurrentSkipListMap za implementaciju neblokirajućeg algoritma koji nam može poslužiti nepromjenjivu snimku podataka čak i ako istodobno više niti ažurira kartu.

Implementacija svih ovih primjera i isječaka koda može se naći u projektu GitHub; ovo je Mavenov projekt, pa bi ga trebalo biti lako uvesti i pokrenuti kakav jest.