Spring Kafka: Implementacja kolejek i Pub-Sub

Wstęp

W tym krótkim samouczku zaczniemy od stworzenia minimalnej konfiguracji — wystarczającej do tworzenia i wykorzystywania wiadomości od brokera Kafki .

Następnie zobaczymy, jak proste jest korzystanie z wiadomości w obu podejściach, takich jak Queues i Pub-Sub .

Następnie omówimy kilka poradników, odpowiedzi na najczęściej zadawane pytania i podstawowe koncepcje — które prawdopodobnie byłyby konieczne w przypadku bardziej złożonych scenariuszy.

Więc bez zbędnych ceregieli, przejdźmy teraz do konfiguracji Kafki Locally.

Szybki start Apache Kafki

Oficjalny przewodnik możemy zobaczyć tutaj:https://kafka.apache.org/quickstart.

Jednak dla zwięzłości możemy wykonać tylko następujące kroki poniżej (wystarczy je śledzić dalej):

1. Pobierz Apache Kafka tutaj
2. Wyodrębnij i włóż do niego:
   2.1 tar -xzf kafka_2.13–3.1.0.tgz
   2.2 cd kafka_2.13–3.1.0
3. Uruchom środowisko Kafka
   3.1 Uruchom Zookeeper: bin/zookeeper-server-start.sh config/zookeeper.properties
   3.2 Uruchom broker Kafka: bin/kafka-server-start.sh config/server.properties

Tworzenie usług producenckich i konsumenckich w Spring Boot

Chociaż możliwe jest tworzenie i konsumowanie wydarzeń za pomocą jednej usługi; jednak żeby było ciekawiej, stworzymy dwa oddzielne projekty — jeden dla producenta i jeden dla konsumenta .

Najpierw dodajmy tę zależność od pom.xml dla obu projektów:

Dodatkowo dla usługi producenta dodajmy tę zależność:

Następnie utwórzmy prosty MessageProducerController , który wysyła wiadomość do brokera Kafka , używając nazwy tematu „temat testowy”:

Przechodząc do obsługi klienta, dla uproszczenia stwórzmy metodę nasłuchiwania w głównej klasie, używając tego samego „tematu testowego”:

Teraz, gdy mamy już podstawową konfigurację minimalną, spróbujmy utworzyć żądanie POST do punktu końcowego /send-message:

curl -X POST http://localhost:8080/send-message \
--header 'Content-Type: application/json' \
--data '{"message": "sample message"}'

Jednak w tym momencie ledwo zarysowaliśmy powierzchnię — nie mieliśmy jeszcze pojęcia, w jaki sposób broker przechowuje lub przetwarza wiadomości. Do tej pory po prostu go spożywaliśmy.

Zanim przejdziemy głębiej, warto zobaczyć, jak proste jest zaimplementowanie powszechnie znanych koncepcji, takich jak kolejki i Pub-Sub w Kafce .

Kolejki i Pub-Sub w Kafce

Rozpoczynając od kolejek, załóżmy, że mamy wielu konsumentów, ale chcemy, aby każda wiadomość była przetwarzana tylko raz. Następnie ustawiamy groupIds konsumentów o tej samej wartości.

Z drugiej strony, w przypadku Pub-Sub, jeśli chcemy przetwarzać komunikat na każdego konsumenta — ustawiamy każdemu z konsumentów groupId z unikalną wartością.

Oto przykład dla obu scenariuszy:

Uwaga : W naszej obecnej konfiguracji nie stworzyliśmy jeszcze konfiguracji dla „tematu testowego”; więc mając tylko jedną partycję, ale dwóch konsumentów z tej samej grupy — jeden z konsumentów będzie bezczynny. Nie poruszyliśmy jeszcze koncepcji przegród, więc o tym później.

Często zadawane pytania / poradniki / koncepcje

1. Jak Kafka Broker przechowuje wiadomości?

Krótka odpowiedź: Poprzez dzienniki.

Broker Kafka tworzy katalog, używając nazwy tematu (i numeru partycji jako konwencji nazewnictwa). Następnie tworzy plik dziennika w tym katalogu.

Aby dowiedzieć się, gdzie znajdują się te logi, możemy sprawdzić plik config/server.properties, a następnie poszukać właściwości „ log.dir ”.

2. Skąd Konsument wie, jaką wiadomość ma przetworzyć dalej?

Krótka odpowiedź: Poprzez offset

Korzystając z naszej prostej implementacji, możemy spróbować sprawdzić przesunięcie za każdym razem, gdy wiadomość jest przetwarzana.

Oto szybki sposób sprawdzenia za pomocą terminala:

bin/kafka-run-class.sh kafka.tools.GetOffsetShell — broker-list localhost:9092 — topic test-topic

Krótka odpowiedź: partycja przechowuje logi. Przypadki użycia: skalowalność (równoległość) i wysoka dostępność (redundancja)

Jak wspomniano wcześniej, broker przechowuje zdarzenia/komunikaty za pośrednictwem dzienników — te dzienniki znajdują się na określonej partycji.

Posiadanie wielu partycji umożliwia wielu konsumentom równoległe przetwarzanie komunikatów.

Aby zapewnić wysoką dostępność, jeśli mamy wiele instancji brokera, możemy ustawić replikację tematu w odniesieniu do liczby instancji brokera — co prowadzi do replikacji partycji w instancjach obserwujących brokera.

4. Jak skonfigurować temat, aby miał wiele partycji i replikacji?

Oto próbka fasoli . Gdzie argumenty dotyczą nazwy tematu, odpowiednio liczby partycji i liczby replikacji.

5. Jak wysyłać/produkować niestandardowe wydarzenia/wiadomości?

W tym celu musimy stworzyć konfigurację producenta. Oto przykład dla większości przypadków użycia przy użyciu JsonSerializer.class:

6. Jak odbierać/konsumować niestandardowe wydarzenia/wiadomości? Jak poradzimy sobie z błędami deserializacji? Co stanie się z offsetem w tym nieudanym scenariuszu?

W tym celu musimy stworzyć konfigurację konsumencką. Oto przykładowa konfiguracja, w której dodaliśmy niestandardowy program obsługi błędów w kontenerze nasłuchiwania Kafka — jednocześnie konfigurując AckMode na MANUAL_IMMEDIATE — aby aktualizować tylko przesunięcie w wywołaniu metody „potwierdzenie”:

Mamy to! Jak zawsze, kod źródłowy jest dostępny na GitHub .