Analiza przypadków użycia Hazelcast w środowisku rozproszonym

Wykorzystanie Hazelcast w projekcie niesie za sobą nowe wyzwania techniczne, z którymi musieliśmy się zmierzyć. W tym artykule postanowiliśmy omówić najciekawsze z nich. Dowiecie się jak użyć Hazelcasta do komunikacji pomiędzy mikroserwisami w oparciu o eventy oraz jak wygląda inicjalizacja Hazelcasta z różnych źródeł danych.

Piotr Dogoda. Backend java developer szczególnie zainteresowany systemami rozproszonymi i reaktywnymi. W BNY Mellon zajmuje się modernizacją wewnętrznych systemów do przesyłania informacji oraz tworzeniem aplikacji. Prywatnie fan koszykówki i cyfrowych gier karcianych.

Mariusz Mikołajczak. Java developer z zamiłowaniem do DevOpsowania. W BNY Mellon zajmuje się implementacją nowych funkcjonalności, integracją z bogatym ekosystemem, a także wprowadzaniem nowinek technologicznych do projektu. Wiecznie głodny nowych wyzwań, zarówno w życiu zawodowym jak i prywatnym. W wolnym czasie eksploruje Bory Dolnośląskie lub przenosi ciężary w przydomowej siłowni.

Użycie Hazelcasta do komunikacji pomiędzy mikroserwisami w oparciu o eventy

Rozproszona architektura oparta o mikroserwisy stała się niepisanym standardem przy budowie bardziej złożonych aplikacji. Oprócz licznych zalet, niesie ona jednak ze sobą pewne wyzwania – za jedno z nich można uznać przesyłanie informacji między serwisami tak, żeby każda instancja serwisu otrzymała daną wiadomość.

Dlaczego warto wybrać Hazelcast?

Kiedy chcemy wysłać wiadomość do naszego serwisu w trybie broadcast, tzn. tak, aby dotarła do wielu odbiorców – w tym przypadku wszystkich instancji naszego serwisu – możemy użyć jednego z gotowych, wyspecjalizowanych rozwiązań jak Apache ZooKeeper czy Spring Cloud Bus. Niosą one jednak ze sobą pewien dodatkowy nakład pracy związany z instalacją i konfiguracją lub wymagają dodatkowych komponentów (w przypadku Spring Cloud Bus potrzebna jest Kafka lub RabbitMQ).

Alternatywnie, jeśli korzystamy już z Hazelcasta, możemy użyć jego systemu eventów. Pozwoli nam to w prosty sposób stworzyć rozwiązanie, które zsynchronizuje mikroserwisy poprzez przesyłanie im wiadomości na zasadzie publish-subscribe. W ten sposób, opublikowaną wiadomość otrzymają wszystkie instancje serwisu, które zasubskrybują dany typ eventu.

Implementacja

1. Publikowanie eventu

Podstawową strukturą danych w Hazelcast są rozproszone mapy. Dostępna jest dla nich funkcjonalność publikowania eventów przy interakcjach takich jak dodawanie, modyfikowanie czy usuwanie elementów.

W tym artykule użyjemy dwóch typów eventów z Hazelcasta: EntryEventType.ADDED i EntryEventType.UPDATED publikowanych odpowiednio przy dodaniu i modyfikacji wpisu w mapie.

Przechodząc do samego publikowania eventów – będziemy potrzebować mapy parametryzowanej typem danych, które chcemy publikować. Wykorzystamy mapę typu <CommandType, String> – pierwszy określający typ wysyłanego polecenia, od którego zależeć będzie w jaki sposób przetworzymy event, a drugi to wartość do przetworzenia.

Mamy tutaj pełną dowolność implementacji. Nic nie stoi na przeszkodzie, aby utworzyć i zarejestrować listenery dla kilku map, sparametryzowanych w inny sposób, w których obsługiwać będziemy eventy różnego typu.

Tworzymy więc taką mapę:

IMap<CommandType, String> eventMap = hazelcastInstance.getMap("event-map");

Teraz, aby opublikować event typu ADDED lub UPDATED dla tej mapy, wywołujemy na niej jedną z metod, która doda lub nadpisze element jak np: put, set lub executeOnKey.

eventMap.set(CommandType.SOME_TYPE, "someExampleValue");

Po wywołaniu tej linii, wszyscy subskrybenci danej mapy otrzymają event typu ADDED, jeśli w mapie nie było wcześniej wartości dla tego klucza, a w przeciwnym wypadku, typu UPDATED.

2. Odbiór eventu

Do przetwarzania eventów publikowanych w pierwszym kroku, użyjemy interfejsu MapListener, a dokładniej jego dwóch rozszerzeń:

• EntryAddedListener<K, V> – otrzymamy wiadomość po dodaniu nowej wartości do mapy (EntryEventType.ADDED),
• EntryUpdatedListener<K, V> – otrzymamy wiadomość po zmodyfikowaniu istniejące wartości w mapie (EntryEventType.UPDATED),

Parametry K i V to oczywiście typ klucza i wartości w mapie.

Przykładowy event listener wykorzystujący powyższe interfejsy będzie więc wyglądał następująco:

public class HazelcastMapEventListener implements
  EntryUpdatedListener<EventType, String>,
  EntryAddedListener<EventType, String> {

 ExampleService exampleService;

 @Override
 public void entryAdded(EntryEvent<CommandType, String> event) {
  processEvent(event);
 }

 @Override
 public void entryUpdated(EntryEvent<CommandType, String> event) {
  processEvent(event);
 }
 
 private void processEvent(EntryEvent<CommandType, String> event) {
  CommandType commandType = event.getKey();
  String eventValue = event.getValue();

  if(CommandType.SOME_TYPE == commandType) {
    exampleService.process(eventValue);
  } else {
    exampleService.processDifferently(eventValue);
  }
 }
}

3. Rejestracja event listenera

Ostatni krok, to zasubskrybowanie event listenera do naszej mapy. Aby to zrobić, każda instancja serwisu zainteresowana odbieraniem wiadomości powinna wywołać metodę addEntryListener na rozproszonej mapie Hazelcasta.

IMap<CommandType, String> eventMap = hazelcastInstance.getMap("event-map");
eventMap.addEntryListener(hazelcastMapEventListener, true);

Podsumowanie

Stworzony w ten sposób system możemy wykorzystać, żeby zsynchronizować rozproszoną aplikację poprzez publikowanie przez Hazelcast informacji, które dotrą do wszystkich instancji serwisu. Pozwala to na dynamiczne zmiany w konfiguracji działających serwisów bez konieczności ponownego ich wdrażania.

Inicjalizacja Hazelcasta z różnych źródeł danych

Czasami chcielibyśmy, aby w momencie inicjalizacji Hazelcasta zostały do niego załadowane dane. Dzięki temu, dostęp nich będzie bardziej wydajny niż gdyby trzeba było je pobrać z persystentnego źródła. W poniższej analizie przyjrzymy się, jak należałoby do tego podejść w sytuacji, gdy jest kilka źródeł danych.

Jak zbudowany jest klaster Hazelcast?

Hazelcast IMDG (In Memory Data Grid) to cache, który używany jest do przechowywania danych w pamięci głównej (RAM), dzięki czemu dostęp do nich jest szybki, około rząd wielkości szybszy niż w przypadku przechowywania ich na dyskach SSD. Dane te są rozprzestrzenione i zreplikowane na współpracujące ze sobą instancje Hazelcasta, które razem tworzą klaster. Każdy node w klastrze przechowuje pewną część danych.

Gdyby z jakiegoś powodu (z powodu awarii serwera, braku prądu itp.) taka instancja została zatrzymana dane zostałyby bezpowrotnie utracone.

Rys.1. Architektura klient-serwer typu embedded. Źródło: smallbusiness.chron.com

Dlatego tworzenie kopii zapasowych i przechowywanie ich na innych instancjach jest niezbędne. Hazelcast umożliwia wykonanie operacji na lokalnej porcji danych – jest to kluczowe i zostanie poruszone w dalszej części artykułu.

Rozproszona mapa – interfejs IMap

IMap, prawdopodobnie najczęściej wykorzystywana struktura danych Hazelcasta, rozszerza Javową klasę ConcurrentMap, a pośrednio także klasę java.util.Map. W odróżnieniu od klasycznej mapy, znanej każdemu developerowi piszącemu w językach działających w oparciu o JVM, IMap jest rozproszoną strukturą danych. Mapa ta zostaje podzielona na partycje, które są równomiernie rozdystrybuowane pomiędzy node’y klastra.

Przy dodawaniu wpisu do mapy, następuje wywołanie funkcji hashującej na kluczu, a na podstawie uzyskanej wartości następuje przyporządkowanie pary klucz-wartość do odpowiedniej partycji.

MapStore, MapLoader – interfejs będzący pomostem pomiędzy Hazelcastem a persystencją

Hazelcast umożliwia wczytywanie/zapisywanie danych z/do rozproszonej mapy np. z/do bazy danych. Aby to osiągnąć, należy zaimplementować odpowiednio interfejs MapLoader lub MapStore. Gdy następuje wywołanie metody IMap.get() i żądana wartość nie znajduje się w mapie, wywoływana jest metoda load() bądź loadAll() z wymienionych wcześniej interfejsów. Wczytana wartość trafia do Hazelcasta, gdzie zostaje zapisana i żyje do momentu aż zostanie jawnie usunięta lub wygaśnie (evicted). Można zauważyć, że wspomniany mechanizm łatwo wykorzystać do inicjalizacji Hazelcasta danymi, które zostały zapisane w sposób trwały.

Kontrakt metod load(), loadAll(), loadAllKeys()

Wygodnym rozwiązaniem jest inicjalizacja pamięci cache danymi pobranymi ze źródła danych, takich jak bazy danych, czy zewnętrzny serwis. Naturalnym sposobem zrealizowania tego jest zaimplementowanie metod: loadAllKeys(), loadAll() oraz load(), które oferują interfejsy MapStore oraz MapLoader. Tryb wczytywania danych określa enumeracja InitialLoadMode – przyjmuje ona dwie wartości: EAGER lub LAZY:

• InitialLoadMode.EAGER oznacza, że wszystkie partycje zostaną załadowane w momencie wystąpienia pierwszej interakcji z mapą (konkretnie w momencie pierwszego wywołania metody HazelcastInstance.getMap(‘’map-name’’)),
• InitialLoadMode.LAZY oznacza, że dane ładowane są partycja po partycji, w momencie wystąpienia pierwszej interakcji z daną partycją.

Metoda loadAllKeys() nie przyjmuje żadnych argumentów oraz zwraca Iterable<K>, gdzie K, to typ wskazujący na klucz mapy. Hazelcast wywołuje tę metodę na jednym ze swoich node’ów, następnie dystrybuuje odczytane klucze pośród wszystkich uczestników klastra. Każdy uczestnik wywołuje wtedy metodę MapLoader.loadAll(keys) i odczytane wpisy zostają umieszczone w IMap’ie.
Wykorzystanie Hazelcast w projekcie niesie za sobą nowe wyzwania techniczne, z którymi musieliśmy się zmierzyć. W tym artykule postanowiliśmy omówić najciekawsze z nich. Dowiecie się jak użyć Hazelcasta do komunikacji pomiędzy mikroserwisami w oparciu o eventy oraz jak wygląda inicjalizacja Hazelcasta z różnych źródeł danych.

Inicjalizacja – baza danych. Runtime – zewnętrzny serwis

Nietrudno wyobrazić sobie następującą sytuację – inicjalizacja cache’a ma odbyć się poprzez załadowanie danych z bazy danych, natomiast po zakończeniu inicjalizacji chcemy dociągać dane z zewnętrznego serwisu. Informacja o tym, że inicjalizacja została zakończona powinna być dostępna dla wszystkich członków klastra, dlatego będzie przechowywana w IMap’ie, która została nazwana initMap. Logika ta powinna być wywoływana jednorazowo w momencie, gdy instancja serwisu zostanie uruchomiona i w pełni zainicjalizowana. Przykładowy fragment kodu wyglądałby następująco:

if (!initMap.get(INIT_MAP_KEY)) {
 mapsToBeLoaded.forEach(this::loadMap);

 initMap.put(INIT_MAP_KEY, true);
 log.info("Finishing up initialization.");
} else {
 log.info("Skipping map initialization as it's been already done.");
}

Gdzie metoda load wygląda następująco:

private void loadMap(String mapName) {
 getHazelcastInstance().getMap(mapName);
 }

Kod.1. Logika wyzwalająca inicjalizację Hazelcasta

Należy zauważyć, że wykorzystanie wspomnianych wcześniej mechanizmów Hazelcastowych (MapStore, MapLoader) zwalnia nas od odpowiedzialności za synchronizację procesu inicjalizacji. Jeśli kilka instancji serwisu wywoła w tym samym czasie metodę HazelcastInstance.getMap() nie spowoduje to dwukrotnego załadowania danych. Zobaczmy, jak wyglądałaby przykładowa implementacja metody loadAll().

@Override
public Map<K, V> loadAll(Collection<K> keys) {
 IMap<String, Boolean> map = HazelcastUtils.initMap();
 boolean loaded = map.get(INIT_MAP_KEY);

 if (loaded) {
  return loadDataFromExternalSource(keys);
 } else {
  return tryToInitializeMap(keys);
 }
}

Kod.2. Implementacja metody loadAll() odziedziczonej z interfejsu MapLoader

Jeśli inicjalizacja została zakończona, czytamy z zewnętrznego serwisu. W przeciwnym przypadku wykonujemy zapytanie do bazy danych celem wyciągnięcia danych.

ReadBackupData oraz BackupCount

Do prawidłowego działania aplikacji należy odpowiednio skonfigurować initMap. Parametr backupCount określa liczbę instancji, na której znajdzie się kopia zapasowa initMap. Wartość ta powinna wynosić n – 1, gdzie n to liczba instancji, które razem tworzą klaster. Dodatkowo, należy nadać parametrowi readBackupData wartość true po to, aby umożliwić Hazelcastowi sięganie do lokalnej kopii mapy initMap. Brak tej konfiguracji (lub sytuacja, gdy konfiguracja jest nieprawidłowa) prędzej bądź później będzie skutkować wyjątkiem o treści: cannot make remote call.

Config config = new Config();
…
config.addMapConfig(
  new MapConfig().setName(INIT_MAP)
   .setBackupCount(numberOfHzInstances - 1)
   .setReadBackupData(true)
   .setMapStoreConfig(
     new MapStoreConfig().setImplementation(
       new InitMapLoader())
      .setEnabled(true)
      .setInitialLoadMode(EAGER))
);

Kod.3. Fragment kodu przedstawiający konfigurację mapy initMap

Należy zwrócić uwagę, że zdefiniowana mapa initMap posiada podpięty MapLoader. Dzięki takiej konfiguracji, za każdym razem gdy aplikacja uruchamiająca Hazelcasta zostanie na nowo wdrożona na środowisko, odbędzie się inicjalizacja z bazy danych (klucz loaded ma domyślnie wartość false). Jego implementacja wygląda następująco:

public class InitMapLoader implements MapLoader<String, Boolean> {

 public final String INIT_MAP_KEY = "loaded";
 private final Map<String, Boolean> map;

 public InitMapLoader() {
  this.map = new HashMap<>();
  map.put(INIT_MAP_KEY, false);
 }

 public Boolean load(String key) {
  return map.get(key);
 }

 public Map&lt;String, Boolean&gt; loadAll(Collection&lt;String&gt; keys) {
  log.info("Initializing init map with loaded key-value.");
  return map.entrySet().stream()
    .filter(e -> keys.contains(e.getKey()))
    .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
 }

 public Iterable<String> loadAllKeys() {
  log.info("Initializing init map with loaded key.");
  return Arrays.asList(INIT_MAP_KEY);
 }
}

Kod.4. Implementacja interfejsu MapLoader dla mapy initMap

Podsumowanie

Zaprezentowany sposób na przełączenie logiki MapLoader/MapStore jest jednym z wielu na rozwiązanie tego problemu. Należy mieć na uwadze, że dostęp do innych struktur danych Hazelcasta z poziomu MapLoader/MapStore może generować potencjalne problemy, np. wystąpienie deadlock’ów. W przedstawionym rozwiązaniu zagrożenie to nie występuje, dzięki czytaniu lokalnej kopii mapy initMap, jednak dzieje się to kosztem spójności danych. W tym przypadku nie jest to aż tak dotkliwe.

Innym sposobem na rozwiązanie problemu mogłoby być wykorzystanie dedykowanej struktury danych com.hazelcast.core.ReplicatedMap, bądź nasłuchiwanie na zmianę wartości mapy initMap i powiązanie tego z logiką zawartą w metodach loadAll()/load().

Oferty pracy z

Zdjęcie główne artykułu pochodzi z unsplash.com.