Kontynuacja tłumaczenia małej książki:
„Zrozumienie brokerów wiadomości”,
autor: Jakub Korab, wydawnictwo: O’Reilly Media, Inc., data wydania: czerwiec 2017, ISBN: 9781492049296.
Poprzednia przetłumaczona część:
ROZDZIAŁ 3
Kafka
Kafka została stworzona w LinkedIn, aby obejść pewne ograniczenia tradycyjnych brokerów wiadomości i uniknąć konieczności konfiguracji wielu brokerów wiadomości dla różnych interakcji „punkt-punkt”, co jest opisane w tej książce w rozdziale „Pionowe i poziome skalowanie” na stronie 28. Scenariusze użycia w LinkedIn opierały się głównie na jednokierunkowym gromadzeniu bardzo dużych ilości danych, takich jak kliknięcia na stronach i logi dostępu, jednocześnie umożliwiając wielokrotne korzystanie z tych danych przez różne systemy, bez wpływu na wydajność producentów lub innych konsumentów. Faktycznie, przyczyną istnienia Kafka jest uzyskanie takiej architektury wymiany wiadomości, jaką opisuje Universal Data Pipeline.
Mając na uwadze ten ostateczny cel, naturalnie powstały także inne wymagania. Kafka musi:
- Być niezwykle szybka
- Zapewniać dużą przepustowość przy obsłudze wiadomości
- Obsługiwać modele „Wydawca-Subskrybent” i „Punkt-Punkt”
- Nie zwalniać przy dodawaniu konsumentów. Na przykład wydajność zarówno kolejek, jak i tematów w ActiveMQ pogarsza się, gdy rośnie liczba konsumentów na adresacie
- Być poziomo skalowalna; jeśli jeden broker, który przechowuje (persists) wiadomości, może to robić tylko z maksymalną prędkością dysku, to dla zwiększenia wydajności ma sens rozważyć więcej niż jedną instancję brokera
- Dzielić dostęp do przechowywania i ponownego wydobywania wiadomości
Aby osiągnąć to wszystko, w Kafka przyjęto architekturę, która przedefiniowała role i obowiązki klientów oraz brokerów wymiany wiadomości. Model JMS jest bardzo ukierunkowany na brokera, który odpowiada za dystrybucję wiadomości, a klienci mają się martwić tylko o wysyłanie i odbieranie wiadomości. Z drugiej strony, Kafka jest ukierunkowana na klienta, przy czym klient przejmuje wiele funkcji tradycyjnego brokera, takich jak sprawiedliwy podział odpowiednich wiadomości wśród konsumentów, w zamian zyskując niezwykle szybki i skalowalny broker. Dla osób, które pracowały z tradycyjnymi systemami wymiany wiadomości, praca z Kafka wymaga fundamentalnych zmian w perspektywie.
To inżynieryjne podejście doprowadziło do stworzenia infrastruktury wymiany wiadomości, która potrafi znacznie zwiększyć przepustowość w porównaniu do zwykłego brokera. Jak się przekonamy, podejście to wiąże się z kompromisami, które oznaczają, że Kafka nie nadaje się do określonych typów obciążeń i oprogramowania systemowego.
Ujednolicona model adresata
Aby spełnić powyższe wymagania, Kafka połączyła wymianę wiadomości typu „publikacja-subskrypcja” i „punkt-punkt” w ramach jednego rodzaju adresata — tematu. To dezorientuje ludzi, którzy pracowali z systemami wymiany wiadomości, gdzie słowo „temat” odnosi się do mechanizmu rozgłaszania, z którego (z tematu) odczyt nie jest niezawodny (is nondurable). Tematy w Kafka należy traktować jako hybrydowy typ adresata, zgodnie z definicją podaną we wprowadzeniu do tej książki.
W pozostałej części tego rozdziału, o ile nie wskażemy inaczej, termin „temat” będzie odnosił się do tematu Kafka.
Aby w pełni zrozumieć, jak zachowują się tematy i jakie gwarancje dają, musimy najpierw przyjrzeć się temu, jak są one zaimplementowane w Kafka.
Każdy temat w Kafka ma swoją własną dziennik.
Producenci wysyłający wiadomości do Kafki zapisują je w tym dzienniku, a konsumenci odczytują z dziennika za pomocą wskaźników, które ciągle przesuwają się do przodu. Okresowo Kafka usuwa najstarsze części dziennika, niezależnie od tego, czy wiadomości w tych częściach zostały odczytane, czy nie. Centralnym elementem projektu Kafki jest to, że broker nie dba o to, czy wiadomości zostały odczytane, czy nie — to odpowiedzialność klienta.
Terminy „dziennik” i „wskaźnik” nie występują w . Te dobrze znane terminy są używane tutaj, aby ułatwić zrozumienie.
Ten model różni się diametralnie od ActiveMQ, gdzie wiadomości ze wszystkich kolejek są przechowywane w jednym dzienniku, a broker oznacza wiadomości jako usunięte po ich odczytaniu.
Teraz zanurzmy się nieco głębiej i przyjrzyjmy się dziennikowi tematu bardziej szczegółowo.
Dziennik Kafki składa się z kilku partycji (). Kafka zapewnia ścisłą kolejność w każdej partycji. Oznacza to, że wiadomości zapisane w partycji w określonej kolejności będą odczytywane w tej samej kolejności. Każda partycja jest realizowana jako cykliczny (rolling) plik dziennika, który zawiera podzbiór (subset) wszystkich wiadomości wysłanych do tematu przez jego producentów. Tworzony temat zawiera domyślnie jedną partycję. Idea partycji to centralny pomysł Kafki na poziome skalowanie.
Rysunek 3-1. Partycje Kafki
Kiedy producent wysyła wiadomość do tematu Kafki, decyduje, do której partycji ją wysłać. Przyjrzymy się temu bardziej szczegółowo później.
Odczyt wiadomości
Klient, który chce odczytać wiadomości, zarządza nazwanym wskaźnikiem, nazywanym grupa konsumentów (consumer group), który wskazuje na offset wiadomości w partycji. Offset to pozycja z rosnącym numerem, która zaczyna się od 0 na początku partycji. Ta grupa konsumentów, na którą odnosi się w API za pomocą zdefiniowanego przez użytkownika identyfikatora group_id, odpowiada jednemu logicznemu konsumentowi lub systemowi.
Większość systemów wykorzystujących wymianę wiadomości odczytuje dane od nadawcy za pomocą wielu instancji i wątków do równoległego przetwarzania wiadomości. Zazwyczaj zatem będzie wiele instancji konsumentów współdzielących tę samą grupę konsumentów.
Problem odczytu można przedstawić w następujący sposób:
- Temat ma kilka partycji
- Z tego samego tematu może korzystać jednocześnie wiele grup konsumentów
- Grupa konsumentów może mieć kilka osobnych instancji
To nie jest trywialny problem 'wielu do wielu'. Aby zrozumieć, jak Kafka radzi sobie z relacjami między grupami konsumentów, instancjami konsumentów i partycjami, rozważmy szereg stopniowo komplikujących się scenariuszy odczytu.
Konsumenci i grupy konsumentów
Weźmy jako punkt wyjścia temat z jedną partycją ().
Rysunek 3-2. Konsument odczytuje z partycji
Gdy instancja konsumenta łączy się ze swoim własnym group_id do tego tematu, przypisywana jest jej partycja do odczytu oraz offset w tej partycji. Pozycja tego offsetu jest konfigurowana w kliencie jako wskaźnik na najnowszą pozycję (najnowsza wiadomość) lub najwcześniejszą pozycję (najstarsza wiadomość). Konsument zadaje (polls) zapytania o wiadomości z tematu, co prowadzi do ich sekwencyjnego odczytu z dziennika.
Pozycja offsetu jest regularnie zatwierdzana z powrotem do Kafki i przechowywana jako wiadomości wewnętrznego tematu _consumer_offsets. Odczytane wiadomości wciąż nie są usuwane, w przeciwieństwie do zwykłego brokera, a klient może przeskoczyć (rewind) offset, aby ponownie przetworzyć już oglądane wiadomości.
Gdy łączy się drugi logiczny konsument, korzystając z innego group_id, zarządza on drugim wskaźnikiem, który nie zależy od pierwszego (). W ten sposób temat Kafki działa jak kolejka, w której istnieje jeden konsument, oraz jak zwykły temat publikator-subskrybent (pub-sub), na który subskrybuje wiele konsumentów, z dodatkową korzyścią, że wszystkie wiadomości są przechowywane i mogą być przetwarzane wiele razy.
Rysunek 3-3. Dwaj konsumenci w różnych grupach konsumentów odczytują z jednej partycji
Konsumenci w grupie konsumentów
Gdy jeden egzemplarz konsumenta odczytuje dane z partycji, ma pełną kontrolę nad wskaźnikiem i przetwarza wiadomości, jak opisano w poprzedniej sekcji.
Jeśli kilka egzemplarzy konsumentów zostało podłączonych z tym samym group_id do tematu z jedną partycją, to egzemplarz, który połączył się ostatni, przejmie kontrolę nad wskaźnikiem i od tego momentu będzie otrzymywał wszystkie wiadomości ().
Rysunek 3-4. Dwaj konsumenci w tej samej grupie konsumentów odczytują z jednej partycji
Ten tryb przetwarzania, w którym liczba egzemplarzy konsumentów przekracza liczbę partycji, można traktować jako formę monopolnego konsumenta. Może to być przydatne, jeśli potrzebujesz „trybu aktywno-pasywnego” (lub „gorącego-ciepłego”) klasteryzacji swoich egzemplarzy konsumentów, chociaż równoległa praca kilku konsumentów („aktywnie-aktywny” lub „gorący-gorący”) jest znacznie bardziej typowa niż konsumenci w trybie oczekiwania.
Takie zachowanie dystrybucji wiadomości, opisane powyżej, może być zaskakujące w porównaniu do tego, jak działa zwykła kolejka JMS. W tym modelu wiadomości wysyłane do kolejki będą równomiernie dystrybuowane między dwoma konsumentami.
Najczęściej, gdy tworzymy kilka egzemplarzy konsumentów, robimy to albo dla równoległego przetwarzania wiadomości, albo dla zwiększenia prędkości odczytu, albo dla zwiększenia odporności procesu odczytu. Ponieważ tylko jeden egzemplarz konsumenta może jednocześnie odczytywać dane z partycji, jak to osiąga się w Kafce?
Jednym ze sposobów realizacji tego celu jest użycie jednego egzemplarza konsumenta do odczytania wszystkich wiadomości i przekazania ich do puli wątków. Chociaż podejście to zwiększa przepustowość przetwarzania, zwiększa również złożoność logiki konsumentów i nic nie robi, aby poprawić odporność systemu odczytu. Jeśli jeden egzemplarz konsumenta zostanie odłączony z powodu awarii zasilania lub podobnego zdarzenia, to odczyt zostaje przerwany.
Kanonizowanym sposobem rozwiązania tego problemu w Kafce jest użycie bowiększej liczby partycji.
Partycjonowanie
Partycje są podstawowym mechanizmem równoległego odczytu i skalowania tematu poza przepustowość pojedynczego brokera. Aby lepiej zrozumieć tę koncepcję, rozważmy sytuację, w której istnieje temat z dwoma partycjami, a do tego tematu subskrybuje jeden konsument ().
Rysunek 3-5. Jeden konsument odczytuje z kilku partycji
W tym scenariuszu konsument ma kontrolę nad wskaźnikami odpowiadającymi jego group_id w obu partycjach i zaczyna odczytywać wiadomości z obu partycji.
Kiedy do tego tematu dodawany jest dodatkowy konsument dla tego samego group_id, Kafka ponownie przydziela (reallocate) jedną z partycji z pierwszego na drugi konsument. Następnie każdy egzemplarz konsumenta będzie odczytywał z jednej partycji tematu ().
Aby zapewnić przetwarzanie wiadomości równolegle w 20 wątkach, potrzebujesz co najmniej 20 partycji. Jeśli partycji jest mniej, pozostaną konsumenty, które nie będą miały nad czym pracować, co zostało wcześniej omówione w kontekście monopolowych konsumentów.
Rysunek 3-6. Dwóch konsumentów w tej samej grupie konsumentów odczytuje z różnych partycji
Ten schemat znacznie zmniejsza złożoność działania brokera Kafka w porównaniu do dystrybucji wiadomości wymaganej do obsługi kolejki JMS. Nie trzeba martwić się o następujące kwestie:
- Który konsument powinien otrzymać następną wiadomość, opierając się na rozdziale okrężnym (round-robin), bieżącej pojemności buforów wstępnego odczytu lub poprzednich wiadomościach (jak w grupach wiadomości JMS).
- Jakie wiadomości zostały wysłane do jakich konsumentów i czy powinny być dostarczone ponownie w przypadku awarii.
Wszystko, co broker Kafka musi zrobić, to kolejno przekazywać wiadomości konsumentowi, gdy ostatni je żąda.
Jednakże wymagania dotyczące równoległego odczytu i ponownego wysyłania nieudanych wiadomości wciąż istnieją — odpowiedzialność za nie przechodzi po prostu z brokera na klienta. Oznacza to, że muszą być uwzględnione w Twoim kodzie.
Wysyłanie wiadomości
Odpowiedzialność za decyzję, do której partycji wysłać wiadomość, spoczywa na producencie tej wiadomości. Aby zrozumieć mechanizm, za pomocą którego to robi się, najpierw należy rozważyć, co tak naprawdę wysyłamy.
Podczas gdy w JMS używamy struktury wiadomości z metadanymi (nagłówkami i właściwościami) oraz ciałem zawierającym ładunek użyteczny (payload), w Kafka wiadomość to para „klucz-wartość”. Ładunek wiadomości jest wysyłany jako wartość (value). Klucz, z drugiej strony, używany jest głównie do partycjonowania i powinien zawierać specyficzny dla logiki biznesowej klucz, aby umieścić powiązane wiadomości w tej samej partycji.
W Rozdziale 2 omawialiśmy scenariusz zakładów online, w którym powiązane zdarzenia muszą być przetwarzane w porządku przez jednego konsumenta:
- Konto użytkownika jest skonfigurowane.
- Pieniądze są wpłacane na konto.
- Składany jest zakład, który wyciąga pieniądze z konta.
Jeśli każde zdarzenie jest wiadomością wysyłaną do tematu, w tym przypadku naturalnym kluczem będzie identyfikator konta.
Gdy wiadomość jest wysyłana za pomocą Kafka Producer API, jest przekazywana do funkcji partycjonowania, która, biorąc pod uwagę wiadomość i aktualny stan klastra Kafka, zwraca identyfikator partycji, do której powinno zostać wysłane to wiadomość. Ta funkcja jest zaimplementowana w Javie przez interfejs Partitioner.
Ten interfejs wygląda następująco:
interface Partitioner {
int partition(String topic,
Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
}Implementacja Partitioner do określenia partycji używa domyślnie algorytmu haszowania klucza (ogólny algorytm haszowania na podstawie klucza) lub algorytmu round-robin, jeśli klucz nie jest podany. Ta wartość domyślna działa dobrze w większości przypadków. Jednak w przyszłości możesz chcieć napisać swoją własną.
Pisanie własnej strategii partycjonowania
Rozważmy przykład, w którym chcesz wysłać metadane razem z ładunkiem użytecznym wiadomości. Ładunek w naszym przykładzie to instrukcja deponowania na konto do gry. Instrukcja to coś, co chcielibyśmy zagwarantować, że nie zostanie zmodyfikowane podczas przesyłania i chcemy mieć pewność, że tylko zaufany zewnętrzny system może zainicjować tę instrukcję. W takim przypadku systemy wysyłające i odbierające uzgadniają użycie podpisu w celu weryfikacji autentyczności wiadomości.
W standardowym JMS po prostu definiujemy właściwość „podpis wiadomości” i dodajemy ją do wiadomości. Jednak Kafka nie zapewnia nam mechanizmu do przesyłania metadanych - tylko klucz i wartość.
Ponieważ wartość to ładunek użyteczny przelewu bankowego (bank transfer payload), którego integralność chcemy zachować, nie mamy innego wyjścia poza zdefiniowanie struktury danych do użycia w kluczu. Zakładając, że potrzebujemy identyfikatora konta do partycjonowania, ponieważ wszystkie wiadomości związane z kontem muszą być przetwarzane w odpowiedniej kolejności, wymyślimy następującą strukturę JSON:
{
"signature": "541661622185851c248b41bf0cea7ad0",
"accountId": "10007865234"
}Ponieważ wartość podpisu będzie się różnić w zależności od ładunku, domyślna strategia haszowania interfejsu Partitioner nie będzie skutecznie grupować powiązanych wiadomości. Dlatego musimy napisać własną strategię, która będzie analizować ten klucz i partycjonować (partition) wartość accountId.
Kafka zawiera sumy kontrolne w celu wykrywania uszkodzenia wiadomości w magazynie i ma pełny zestaw funkcji zabezpieczeń. Mimo to czasami pojawiają się specyficzne wymagania branżowe, jak to wymienione powyżej.
Niestandardowa strategia partycjonowania musi gwarantować, że wszystkie powiązane wiadomości trafią do jednej partycji. Choć wydaje się to proste, wymaganie to może być skomplikowane z uwagi na znaczenie uporządkowania powiązanych wiadomości i to, jak sztywna jest liczba partycji w temacie.
Liczba partycji w temacie może zmieniać się w czasie, ponieważ można je dodać, jeśli ruch przekracza początkowe oczekiwania. W ten sposób klucze wiadomości mogą być związane z partycją, do której pierwotnie zostały wysłane, co sugeruje część stanu, który powinien być rozdzielony między instancje producenta.
Innym czynnikiem, który należy wziąć pod uwagę, jest sprawiedliwość rozkładu wiadomości między partycje. Z reguły klucze nie są rozdzielane równomiernie w wiadomościach, a funkcje haszujące nie gwarantują sprawiedliwego rozkładu wiadomości dla małego zestawu kluczy.
Ważne jest, aby zwrócić uwagę na to, że niezależnie od tego, jak postanowisz podzielić wiadomości, separator może być konieczny do ponownego użycia.
Rozważmy wymóg replikacji danych pomiędzy klastrami Kafka w różnych lokalizacjach geograficznych. W tym celu Kafka dostarcza narzędzie wiersza poleceń o nazwie MirrorMaker, które służy do odczytywania wiadomości z jednego klastra i przesyłania ich do drugiego.
MirrorMaker musi rozumieć klucze replikowanego tematu, aby zachować względną kolejność wiadomości podczas replikacji między klastrami, ponieważ liczba partycji dla tego tematu może być różna w obu klastrach.
Własne strategie partycjonowania występują stosunkowo rzadko, ponieważ domyślne metody, takie jak haszowanie lub cykliczne przypisanie, skutecznie działają w większości scenariuszy. Jednak jeśli potrzebujesz ścisłych gwarancji porządkowania lub musisz wydobyć metadane z ładunków, partycjonowanie to coś, na co warto zwrócić większą uwagę.
Zalety skalowalności i wydajności Kafki wynikają z przeniesienia niektórych obowiązków tradycyjnego brokera na klienta. W takim przypadku podejmuje się decyzję o rozdzieleniu potencjalnie związanych wiadomości między kilkoma konsumującymi, działającymi równolegle.
Brokerzy JMS również muszą radzić sobie z takimi wymaganiami. Co ciekawe, mechanizm przesyłania związanych wiadomości do tego samego konsumenta, zrealizowany za pomocą grup wiadomości JMS (rodzaj strategii balansowania obciążenia typu sticky load balancing (SLB)), wymaga również, aby nadawca oznaczał wiadomości jako powiązane. W przypadku JMS broker odpowiada za przesyłanie tej grupy związanych wiadomości do jednego z wielu konsumentów oraz za przekazywanie praw własności grupy, jeśli konsument się rozłączy.
Ustalenia dotyczące producenta
Partycjonowanie to nie jedyne, co należy wziąć pod uwagę przy wysyłaniu wiadomości. Przyjrzyjmy się metodom send() klasy Producer w API Java:
Future send(ProducerRecord record);
Future send(ProducerRecord record, Callback callback);Należy od razu zauważyć, że obie metody zwracają Future, co wskazuje, że operacja wysyłania nie jest wykonywana natychmiastowo. W rezultacie wiadomość (ProducerRecord) jest zapisywana w buforze wysyłania dla każdej aktywnej partycji i przekazywana brokerowi w tle przez bibliotekę klienta Kafka. Choć czyni to pracę niezwykle szybką, oznacza to, że źle napisane aplikacje mogą tracić wiadomości, jeśli ich proces zostanie zatrzymany.
Jak zawsze, istnieje sposób, aby uczynić operację wysyłania bardziej niezawodną kosztem wydajności. Rozmiar tego bufora można ustawić na 0, a wątek aplikacji wysyłającej będzie musiał poczekać, aż przesyłanie wiadomości do brokera zostanie zakończone w następujący sposób:
RecordMetadata metadata = producer.send(record).get();Jeszcze raz o odczytywaniu wiadomości
Odczytywanie wiadomości ma dodatkowe złożoności, nad którymi trzeba się zastanowić. W przeciwieństwie do API JMS, które może uruchomić nasłuchiwacz wiadomości (message listener) w odpowiedzi na otrzymanie wiadomości, interfejs Consumer Kafka tylko polluje (polling). Przyjrzyjmy się bliżej metodzie poll (), używanej do tego celu:
ConsumerRecords poll(long timeout);Wartością zwracaną przez metodę jest struktura kontenerowa zawierająca wiele obiektów ConsumerRecord z potencjalnie wielu partycji. ConsumerRecord jest sam w sobie obiektem przechowującym parę klucz-wartość z odpowiednimi metadanymi, takimi jak partycja, z której został pobrany.
Jak omówiono w Rozdziale 2, musimy nieustannie pamiętać, co dzieje się z wiadomościami po ich pomyślnym lub niepomyślnym przetworzeniu, na przykład jeśli klient nie może przetworzyć wiadomości lub jeśli przerywa działanie. W JMS było to obsługiwane za pomocą trybu potwierdzenia (acknowledgement mode). Broker albo usunięje pomyślnie przetworzoną wiadomość, albo ponownie dostarcza nieprzetworzoną lub zepsutą (pod warunkiem, że użyto transakcji).
Kafka działa zupełnie inaczej. Wiadomości nie są usuwane w brokerze po odczytaniu, a odpowiedzialność za to, co dzieje się w przypadku awarii, spoczywa na samym kodzie odczytującym.
Jak już wspomniano, grupa konsumentów jest związana z przesunięciem w dzienniku. Pozycja w dzienniku związana z tym przesunięciem odpowiada następnej wiadomości, która zostanie wydana w odpowiedzi na poll ()Decydujące znaczenie w trakcie czytania ma moment, w którym to przesunięcie wzrasta.
Wracając do modelu czytania, rozważanego wcześniej, przetwarzanie komunikatu składa się z trzech etapów:
- Wyodrębnić komunikat do przeczytania.
- Przetworzyć komunikat.
- Potwierdzić komunikat.
Konsument Kafka jest wyposażony w opcję konfiguracyjną enable.auto.commit. To często używane ustawienie domyślne, jak to zwykle bywa w przypadku ustawień zawierających słowo „auto”.
Do wersji Kafka 0.10 klient, który używał tego parametru, przesyłał przesunięcie ostatnio przeczytanego komunikatu przy następnym wywołaniu poll () po przetworzeniu. Oznaczało to, że wszelkie komunikaty, które już zostały wyodrębnione (fetched), mogły być przetwarzane ponownie, jeśli klient je już przetworzył, ale został niespodziewanie zniszczony przed wywołaniem poll (). Ponieważ broker nie przechowuje żadnego stanu dotyczącego tego, ile razy komunikat został przeczytany, następny konsument, który wyodrębnia ten komunikat, nie będzie wiedział, że wydarzyło się coś złego. To zachowanie było pseudo-transakcyjne. Przesunięcie było zatwierdzane tylko w przypadku pomyślnego przetworzenia komunikatu, ale jeśli klient przerywał działanie, broker ponownie przesyłał ten sam komunikat innemu klientowi. Takie zachowanie odpowiadało gwarancji dostarczania komunikatów „co najmniej raz«.
W wersji Kafka 0.10 kod klienta został zmieniony w taki sposób, że zatwierdzanie stało się okresowo uruchamiane przez bibliotekę klienta, zgodnie z ustawieniem auto.commit.interval.ms. To zachowanie znajduje się gdzieś pomiędzy trybami JMS AUTO_ACKNOWLEDGE a DUPS_OK_ACKNOWLEDGE. Przy użyciu automatycznego zatwierdzenia komunikaty mogły być potwierdzane niezależnie od tego, czy zostały faktycznie przetworzone — mogło to wystąpić w przypadku wolnego konsumenta. Jeśli konsument przerywał działanie, komunikaty były wyodrębniane przez kolejnego konsumenta, zaczynając od zatwierdzonej pozycji, co mogło prowadzić do pominięcia komunikatu. W takim przypadku Kafka nie traciła komunikatów, kod odczytujący po prostu ich nie przetwarzał.
Ten tryb ma te same perspektywy, co w wersji 0.9: komunikaty mogą być przetwarzane, ale w przypadku awarii przesunięcie może nie być zatwierdzone, co potencjalnie może prowadzić do podwójnego dostarczenia. Im więcej komunikatów wyodrębniasz podczas wykonywania poll (), tym większy jest ten problem.
Jak omówiono w rozdziale „Odczytywanie wiadomości z kolejki” na str. 21, w systemie wymiany wiadomości nie ma pojęcia jednorazowej dostawy wiadomości, jeśli uwzględnić tryby awarii.
W Kafka istnieją dwa sposoby na zarejestrowanie (zatwierdzenie) przesunięcia (offsetu): automatycznie i ręcznie. W obu przypadkach wiadomości mogą być przetwarzane wielokrotnie, jeśli wiadomość została przetworzona, ale wystąpił błąd przed zatwierdzeniem. Możesz także w ogóle nie przetwarzać wiadomości, jeśli zatwierdzenie nastąpiło w tle, a twój kod zakończył się zanim zaczął przetwarzanie (możliwe w Kafka 0.9 i wcześniejszych wersjach).
Można ręcznie zarządzać procesem zatwierdzania offsetu w API konsumenta Kafka, ustawiając parametr enable.auto.commit na wartość false i jawnie wywołując jedną z następujących metod:
void commitSync();
void commitAsync();Jeśli chcesz przetworzyć wiadomość „przynajmniej raz”, musisz ręcznie zatwierdzić offset za pomocą commitSync (), wykonując tę komendę zaraz po przetworzeniu wiadomości.
Te metody nie pozwalają na potwierdzanie (acknowledged) wiadomości, zanim zostaną przetworzone, ale nic nie robią, aby uniknąć potencjalnego podwójnego przetworzenia, tworząc jednocześnie wrażenie transakcyjności. W Kafka nie ma transakcji. Klient nie ma możliwości wykonania następujących czynności:
- Automatycznie cofnąć (roll back) nieudane wiadomości. Konsumenci muszą samodzielnie radzić sobie z wyjątkami wynikającymi z problematycznych ładunków i rozłączeń backendu, ponieważ nie mogą polegać na ponownej dostawie wiadomości przez brokera.
- Wysyłać wiadomości do kilku tematów w jednej atomowej operacji. Jak wkrótce zobaczymy, kontrola nad różnymi tematami i partycjami może znajdować się na różnych maszynach w klastrze Kafka, które nie koordynują transakcji przy wysyłaniu. W momencie pisania tego artykułu wykonano pewną pracę, aby to umożliwić dzięki KIP-98.
- Powiązać odczyt jednej wiadomości z jednego tematu z wysłaniem innej wiadomości do innego tematu. Jeszcze raz, architektura Kafka opiera się na wielu niezależnych maszynach działających jak jeden szynowy system i nie podejmuje się żadnych prób ukrycia tego. Na przykład nie ma komponentów API, które pozwalałyby powiązać Konsument i Producent w transakcji. W JMS zapewnia to obiekt Sesja, z którego są tworzone MessageProducers i MessageConsumers.
Jeśli nie możemy polegać na transakcjach, jak możemy zapewnić semantykę, bliższą tej, którą oferują tradycyjne systemy wymiany wiadomości?
Jeśli istnieje prawdopodobieństwo, że offset konsumenta może wzrosnąć zanim wiadomość zostanie przetworzona, np. podczas awarii konsumenta, to konsument nie ma sposobu, aby dowiedzieć się, czy jego grupa konsumentów pominęła wiadomości, gdy przypisano jej partycję. W ten sposób jedna ze strategii polega na przewinięciu offsetu do poprzedniej pozycji. API konsumenta Kafka oferuje następujące metody do tego:
void seek(TopicPartition partition, long offset);
void seekToBeginning(Collection partitions); Metoda seek () może być używana z metodą
offsetsForTimes (Map timestampsToSearch) aby przewinąć do stanu w określonym momencie w przeszłości.
Niejawnie, użycie tego podejścia oznacza, że jest bardzo prawdopodobne, iż niektóre wiadomości, które zostały wcześniej przetworzone, zostaną odczytane i przetworzone ponownie. Aby tego uniknąć, możemy zastosować idempotentne odczyty, jak opisano w Rozdziale 4, aby śledzić wcześniej wyświetlane wiadomości i wykluczać duplikaty.
Alternatywnie, kod twojego konsumenta może być prosty, jeśli dopuszczalna jest utrata lub duplikacja wiadomości. Kiedy rozważamy przypadki użycia, dla których zazwyczaj wykorzystywana jest Kafka, takie jak przetwarzanie zdarzeń logów, metryk, śledzenie kliknięć itd., rozumiemy, że utrata pojedynczych wiadomości raczej nie wpłynie znacząco na otaczające aplikacje. W takich przypadkach wartości domyślne są jak najbardziej akceptowalne. Z drugiej strony, jeśli twoja aplikacja musi przesyłać płatności, powinieneś starannie dbać o każdą pojedynczą wiadomość. Wszystko sprowadza się do kontekstu.
Osobiste obserwacje pokazują, że wraz ze wzrostem intensywności wiadomości, wartość każdej pojedynczej wiadomości maleje. Wiadomości dużych objętości stają się zazwyczaj cenne, jeśli są rozpatrywane w formie skonsolidowanej.
Wysoka dostępność (High Availability)
Podejście Kafka do zapewnienia wysokiej dostępności znacznie różni się od podejścia ActiveMQ. Kafka została zaprojektowana na bazie poziomo skalowalnych klastrów, w których wszystkie instancje brokera jednocześnie przyjmują i przekazują wiadomości.
Klastor Kafka składa się z kilku instancji brokera działających na różnych serwerach. Kafka została zaprojektowana do działania na standardowym sprzęcie autonomicznym, gdzie każdy węzeł ma własne, dedykowane miejsce do przechowywania. Użycie sieciowych magazynów (SAN) nie jest zalecane, ponieważ wiele węzłów obliczeniowych może konkurować o czasowe interwały przechowywania i powodować konflikty.Ye interwały przechowywania i tworzyć konflikty.
Kafka to system ciągle włączony. Wielu dużych użytkowników Kafka nigdy nie wyłącza swoich klastrów, a oprogramowanie zapewnia aktualizację poprzez sekwencyjne ponowne uruchamianie. Osiąga się to przez zapewnienie zgodności z poprzednią wersją dla wiadomości i interakcji między brokerami.
Brokerzy są podłączeni do klastra serwerów , który działa jak rejestr danych konfiguracyjnych i jest używany do koordynowania ról każdego brokera. ZooKeeper sam w sobie jest rozproszonym systemem, który zapewnia wysoką dostępność poprzez replikację informacji poprzez ustanowienie kwarum.
W podstawowym przypadku temat jest tworzony w klastrze Kafka z następującymi właściwościami:
- Liczba partycji. Jak omówiono wcześniej, dokładna wartość używana tutaj zależy od pożądanego poziomu równoległego odczytu.
- Współczynnik (czynnik) replikacji określa, ile instancji brokera w klastrze powinno przechowywać dzienniki dla tej partycji.
Korzystając z ZooKeepers do koordynacji, Kafka stara się sprawiedliwie rozdzielić nowe partycje między brokerami w klastrze. Robi to jeden z brokerów, który pełni rolę Kontrolera.
W czasie wykonania dla każdej partycji tematu Kontroler przydziela brokerowi role lidera (leader, master, wiodący) oraz naśladowców (followers, slaves, podrzędnych). Broker, pełniący rolę lidera dla danej partycji, odpowiada za przyjmowanie wszystkich wiadomości wysyłanych mu przez producentów i dystrybucję wiadomości do konsumentów. Gdy wiadomości są wysyłane do partycji tematu, są replikowane na wszystkich węzłach brokera, pełniących rolę naśladowców dla tej partycji. Każdy węzeł, który zawiera dzienniki dla partycji, nazywany jest repliką. Broker może pełnić rolę lidera dla niektórych partycji i rolę naśladowcy dla innych.
Naśladowca, który zawiera wszystkie wiadomości przechowywane u lidera, nazywany jest synchronizowaną repliką (replica, która jest w zsynchronizowanym stanie, in-sync replica). Jeśli broker, pełniący rolę lidera dla partycji, zostanie wyłączony, każdy broker, który jest w aktualizowanym lub zsynchronizowanym stanie dla tej partycji, może przejąć rolę lidera. To niezwykle odporna konstrukcja.
Częścią konfiguracji producenta jest parametr acks, który określa, ile replik musi potwierdzić (acknowledge) otrzymanie wiadomości, zanim strumień aplikacji będzie kontynuował wysyłanie: 0, 1 lub wszystkie. Jeśli wartość wszystkojest ustawiona, to po otrzymaniu wiadomości lider wyśle potwierdzenie (confirmation) z powrotem do producenta, jak tylko otrzyma potwierdzenia (acknowledgements) zapisu od kilku replik (w tym od samego siebie), określonych ustawieniem tematu min.insync.replicas (domyślnie 1). Jeśli wiadomość nie może być pomyślnie replikowana, producent zgłosi wyjątek dla aplikacji (NotEnoughReplicas lub NotEnoughReplicasAfterAppend).
W typowej konfiguracji tworzony jest temat z współczynnikiem replikacji 3 (1 lider, 2 naśladowców dla każdej partycji) i parametr min.insync.replicas jest ustawiony na wartość 2. W takim przypadku klaster dopuszcza, aby jeden z brokerów zarządzających partycją tematu mógł zostać wyłączony bez wpływu na aplikacje klienckie.
To prowadzi nas z powrotem do znanego już kompromisu między wydajnością a niezawodnością. Replikacja odbywa się z dodatkowym czasem oczekiwania na potwierdzenia (acknowledgments) od naśladowców. Mimo to, ponieważ odbywa się równolegle, replikacja, co najmniej na trzech węzłach, ma tę samą wydajność jak na dwóch (ignorując wzrost wykorzystania przepustowości sieci).
Korzystając z tego schematu replikacji, Kafka zręcznie unika konieczności zapewnienia fizycznego zapisu każdej wiadomości na dysku za pomocą operacji sync (). Każda wiadomość wysłana przez producenta będzie zapisywana w dzienniku partycji, ale, jak omówiono w Rozdziale 2, zapis do pliku początkowo odbywa się w buforze systemu operacyjnego. Jeśli ta wiadomość zostanie zreplikowana na inną instancję Kafki i znajduje się w jej pamięci, utrata lidera nie oznacza, że sama wiadomość została utracona – może ją przejąć zsynchronizowana replika.
Rezygnacja z konieczności wykonania operacji sync () oznacza, że Kafka może przyjmować wiadomości z prędkością, z jaką może je zapisywać w pamięci. I odwrotnie, im dłużej można unikać zrzucania (flushing) pamięci na dysk, tym lepiej. Z tego powodu nie jest rzadkością, że brokerom Kafki przydziela się 64 GB pamięci lub więcej. Takie wykorzystanie pamięci oznacza, że jedna instancja Kafki może łatwo działać z prędkościami wielokrotnie przekraczającymi tradycyjnego brokera wiadomości.
Kafka może być również skonfigurowana do stosowania operacji sync () na pakietach wiadomości. Ponieważ wszystko w Kafce jest zorientowane na pracę z pakietami, w rzeczywistości działa to dość dobrze w wielu scenariuszach użycia i jest użytecznym narzędziem dla użytkowników, którzy wymagają bardzo silnych gwarancji. Większość czystej wydajności Kafki związana jest z wiadomościami, które są wysyłane do brokera w postaci pakietów, oraz z tym, że te wiadomości są odczytywane z brokera sekwencyjnie w blokach za pomocą operacji (operacjami, w trakcie których nie wykonywana jest operacja kopiowania danych z jednej przestrzeni pamięci do drugiej). Ostatnie jest dużą zaletą z punktu widzenia wydajności i zasobów i jest możliwe tylko dzięki zastosowaniu leżącej u podstaw struktury danych dziennika, określającej schemat partycji.
W klastrze Kafka możliwa jest znacznie wyższa wydajność niż przy użyciu jednego brokera Kafka, ponieważ partycje tematu mogą być poziomo skalowane na wielu oddzielnych maszynach.
Podsumowanie
W tej części rozdziału omówiliśmy, jak architektura Kafka reinterpretacja relacji między klientami a brokerami, aby zapewnić niezwykle niezawodny kanał wymiany wiadomości, o przepustowości wielokrotnie wyższej niż standardowy broker wiadomości. Dyskutowaliśmy o funkcjonalności, którą wykorzystuje do osiągnięcia tego celu oraz krótko przedstawiliśmy architekturę aplikacji zapewniających tę funkcjonalność. W następnej części rozdziału omówimy powszechne problemy, które muszą rozwiązywać aplikacje oparte na wymianie wiadomości, oraz przedyskutujemy strategie ich rozwiązywania. Zakończymy rozdział, wskazując, jak rozumieć technologie wymiany wiadomości jako całość, aby móc ocenić ich przydatność do twoich scenariuszy użycia.
Poprzednia przetłumaczona część:
Tłumaczenie wykonano:
Ciąg dalszy nastąpi…
Tylko zarejestrowani użytkownicy mogą brać udział w ankiecie. , proszę.
Czy używasz Kafka w swojej organizacji?
Tak
Nie
Kiedyś używane, teraz nie
Planujemy używać
38 użytkowników zagłosowało. 8 użytkowników wstrzymało się od głosu.
Źródło: habr.com
