Mamy sytuację, w której mam do czynienia z ogromnym napływem zdarzeń przychodzących na nasz serwer, średnio przy około 1000 zdarzeń na sekundę (szczyt może wynosić ~ 2000).

Problem

Nasz system jest hostowany na Heroku i używa stosunkowo drogiej bazy danych Heroku Postgres , która pozwala na maksymalnie 500 połączeń DB. Korzystamy z pul połączeń do łączenia się z serwera do bazy danych.

Zdarzenia przychodzą szybciej niż pula połączeń DB może obsłużyć

Problemem jest to, że zdarzenia przychodzą szybciej niż pula połączeń jest w stanie obsłużyć. Zanim jedno połączenie zakończy połączenie sieciowe z serwera do bazy danych, aby można je było zwolnić z powrotem do puli, npojawiły się więcej niż dodatkowe zdarzenia.

W końcu zdarzenia gromadzą się, czekając na zapisanie, a ponieważ w puli nie ma dostępnych połączeń, przekroczą limit czasu i cały system przestanie działać.

Rozwiązaliśmy sytuację awaryjną, emitując obrażające zdarzenia o wysokiej częstotliwości w wolniejszym tempie od klientów, ale nadal chcemy wiedzieć, jak radzić sobie z tymi scenariuszami w przypadku, gdy musimy poradzić sobie z tymi zdarzeniami o wysokiej częstotliwości.

Ograniczenia

Inni klienci mogą chcieć czytać zdarzenia jednocześnie

Inni klienci stale żądają odczytania wszystkich zdarzeń z określonym kluczem, nawet jeśli nie są jeszcze zapisane w bazie danych.

Klient może GET api/v1/events?clientId=1wysłać zapytanie i uzyskać wszystkie zdarzenia wysłane przez klienta 1, nawet jeśli te zdarzenia nie zostały jeszcze zapisane w bazie danych.

Czy są jakieś „klasowe” przykłady, jak sobie z tym poradzić?

Możliwe rozwiązania

Kolejkuj wydarzenia na naszym serwerze

Możemy kolejkować zdarzenia na serwerze (z kolejką o maksymalnej współbieżności wynoszącej 400, aby pula połączeń się nie wyczerpała).

To zły pomysł, ponieważ:

• Zużyje dostępną pamięć serwera. Ułożone w kolejce zdarzenia zużyją ogromne ilości pamięci RAM.
• Nasze serwery restartują się raz na 24 godziny . Jest to twardy limit narzucony przez Heroku. Serwer może się zrestartować, gdy zdarzenia są kolejkowane, co powoduje utratę tych zdarzeń.
• Wprowadza stan na serwerze, co szkodzi skalowalności. Jeśli mamy konfigurację z wieloma serwerami, a klient chce odczytać wszystkie kolejkowane i zapisane zdarzenia, nie będziemy wiedzieć, na którym serwerze kolejkowane zdarzenia są aktywne.

Użyj osobnej kolejki komunikatów

Zakładam, że moglibyśmy użyć kolejki komunikatów (np. RabbitMQ ?), W której pompujemy wiadomości w niej, a na drugim końcu jest inny serwer, który zajmuje się tylko zapisywaniem zdarzeń na DB.

Nie jestem pewien, czy kolejki komunikatów umożliwiają odpytywanie zakolejkowanych zdarzeń (które nie zostały jeszcze zapisane), więc jeśli inny klient chce odczytać wiadomości innego klienta, mogę po prostu pobrać zapisane wiadomości z bazy danych i oczekujące wiadomości z kolejki i łączymy je ze sobą, dzięki czemu mogę wysłać je z powrotem do klienta żądania odczytu.

Korzystaj z wielu baz danych, z których każda zapisuje część wiadomości za pomocą centralnego serwera koordynującego DB, aby nimi zarządzać

Innym rozwiązaniem, które mamy, jest wykorzystanie wielu baz danych z centralnym „koordynatorem DB / modułem równoważącym obciążenie”. Po otrzymaniu zdarzenia koordynator wybierze jedną z baz danych, do których napisze wiadomość. Powinno to pozwolić nam korzystać z wielu baz danych Heroku, zwiększając w ten sposób limit połączeń do 500 x liczby baz danych.

Po zapytaniu dotyczącym odczytu koordynator może SELECTwysyłać zapytania do każdej bazy danych, scalać wszystkie wyniki i odsyłać je z powrotem do klienta, który zażądał odczytu.

To zły pomysł, ponieważ:

• Ten pomysł brzmi jak ... hmm ... nadmierna inżynieria? Byłby to również koszmar do zarządzania (kopie zapasowe itp.). Kompilacja i utrzymanie jest skomplikowane i chyba, że ​​jest to absolutnie konieczne, brzmi jak naruszenie zasad KISS .
• Poświęca spójność . Robienie transakcji na wielu bazach danych jest nie do przyjęcia, jeśli pójdziemy z tym pomysłem.

Strumień wejściowy

Nie jest jasne, czy 1000 zdarzeń na sekundę reprezentuje wartości szczytowe, czy jest to ciągłe obciążenie:

• jeśli jest to szczyt, można użyć kolejki komunikatów jako bufora, aby rozłożyć obciążenie serwera DB na dłuższy czas;
• jeśli jest to stałe obciążenie, sama kolejka komunikatów nie jest wystarczająca, ponieważ serwer DB nigdy nie będzie w stanie go dogonić. Następnie musisz pomyśleć o rozproszonej bazie danych.

Proponowane rozwiązanie

Intuicyjnie w obu przypadkach wybrałbym strumień zdarzeń oparty na Kafce :

• Wszystkie wydarzenia są systematycznie publikowane na temat kafki
• Konsument zasubskrybuje wydarzenia i zapisze je w bazie danych.
• Procesor zapytań obsłuży żądania klientów i prześle zapytanie do bazy danych.

Jest to wysoce skalowalne na wszystkich poziomach:

• Jeśli wąskim gardłem jest serwer DB, po prostu dodaj kilku konsumentów. Każdy może subskrybować ten temat i pisać na innym serwerze DB. Jeśli jednak dystrybucja odbywa się losowo na serwerach DB, procesor zapytań nie będzie w stanie przewidzieć, że serwer DB podejmie, i będzie musiał wysłać zapytanie do kilku serwerów DB. Może to prowadzić do nowego wąskiego gardła po stronie zapytania.
• Schemat dystrybucji DB można zatem przewidzieć, organizując strumień zdarzeń na kilka tematów (na przykład, używając grup kluczy lub właściwości, aby podzielić DB zgodnie z przewidywalną logiką).
• Jeśli jeden serwer komunikatów nie jest wystarczający, aby obsłużyć rosnącą liczbę zdarzeń wejściowych, można dodać partycje kafka w celu dystrybucji tematów kafka na kilka serwerów fizycznych.

Oferowanie klientom zdarzeń, które nie zostały jeszcze zapisane w bazie danych

Chcesz, aby Twoi klienci mogli uzyskać dostęp również do informacji, które są jeszcze w potoku i nie zostały jeszcze zapisane w bazie danych. To jest trochę bardziej delikatne.

Opcja 1: Używanie pamięci podręcznej w celu uzupełnienia zapytań db

Nie analizowałem dogłębnie, ale pierwszym pomysłem, jaki przychodzi mi na myśl, byłoby uczynienie procesora (ów) zapytań konsumentem (-ami) tematów kafka, ale w innej grupie konsumentów kafka . Procesor żądań otrzyma wówczas wszystkie wiadomości, które otrzyma program zapisujący DB, ale niezależnie. Może wtedy przechowywać je w lokalnej pamięci podręcznej. Zapytania byłyby następnie uruchamiane na buforze DB + (+ eliminacja duplikatów).

Projekt wyglądałby następująco:

Skalowalność tej warstwy zapytań można osiągnąć, dodając więcej procesorów zapytań (każdy w osobnej grupie konsumentów).

Opcja 2: zaprojektuj podwójne API

Lepszym podejściem IMHO byłoby zaoferowanie podwójnego API (skorzystaj z mechanizmu oddzielnej grupy konsumentów):

• API zapytania do uzyskiwania dostępu do zdarzeń w bazie danych i / lub dokonywania analiz
• interfejs API przesyłania strumieniowego, który przekazuje wiadomości bezpośrednio z tematu

Zaletą jest to, że pozwalasz klientowi decydować, co jest interesujące. Pozwoli to uniknąć systematycznego łączenia danych DB ze świeżo zainkasowanymi danymi, gdy klient jest zainteresowany tylko nowymi przychodzącymi zdarzeniami. Jeśli naprawdę potrzebne jest delikatne połączenie świeżych i zarchiwizowanych wydarzeń, klient musiałby je zorganizować.

Warianty

Zaproponowałem kafkę, ponieważ jest przeznaczony do bardzo dużych woluminów z trwałymi komunikatami, aby w razie potrzeby można było ponownie uruchomić serwery.

Możesz zbudować podobną architekturę za pomocą RabbitMQ. Jednak jeśli potrzebujesz trwałych kolejek, może to obniżyć wydajność . Ponadto, o ile mi wiadomo, jedynym sposobem na osiągnięcie równoległego zużycia tych samych wiadomości przez kilka czytników (np. Writer + cache) w RabbitMQ jest klonowanie kolejek . Tak więc wyższa skalowalność może mieć wyższą cenę.

Domyślam się, że musisz dokładniej zbadać podejście, które odrzuciłeś

• Kolejkuj wydarzenia na naszym serwerze

Moją sugestią byłoby zacząć od przeczytania różnych artykułów opublikowanych na temat architektury LMAX . Udało im się wykonać wsadowe przetwarzanie dużych ilości dla ich przypadku użycia, i może być możliwe, aby twoje kompromisy wyglądały bardziej jak ich.

Możesz także sprawdzić, czy możesz usunąć odczyty z drogi - najlepiej byłoby móc skalować je niezależnie od zapisów. Może to oznaczać zajrzenie do CQRS (segregacja odpowiedzialności za zapytania).

Serwer może się zrestartować, gdy zdarzenia są kolejkowane, co powoduje utratę tych zdarzeń.

W systemie rozproszonym myślę, że możesz być całkiem pewny, że wiadomości zostaną zgubione. Możesz być w stanie złagodzić część tego wpływu, rozważając bariery sekwencji (na przykład - upewniając się, że nastąpi zapis do trwałej pamięci - zanim wydarzenie zostanie udostępnione poza systemem).

• Korzystaj z wielu baz danych, z których każda zapisuje część wiadomości za pomocą centralnego serwera koordynującego DB, aby nimi zarządzać

Może - Bardziej prawdopodobne byłoby przyjrzenie się granicom biznesowym, aby sprawdzić, czy istnieją naturalne miejsca na odłamki danych.

Czy są przypadki, w których utrata danych jest akceptowalnym kompromisem?

Cóż, przypuszczam, że może być, ale nie tam chodziłem. Chodzi o to, że projekt powinien mieć w sobie solidność wymaganą do postępu w obliczu utraty wiadomości.

Często wygląda to na model ściągania z powiadomieniami. Dostawca zapisuje wiadomości w zamówionym trwałym sklepie. Konsument wyciąga wiadomości ze sklepu, śledząc swój własny znak wysokiej wody. Powiadomienia wypychane są używane jako urządzenie zmniejszające opóźnienia - ale w przypadku zgubienia powiadomienia wiadomość jest nadal pobierana (ostatecznie), ponieważ konsument korzysta z regularnego harmonogramu (różnica polega na tym, że jeśli powiadomienie zostanie odebrane, ściąganie nastąpi wcześniej ).

Zobacz: Wiarygodne przesyłanie wiadomości bez rozproszonych transakcji autorstwa Udi Dahana (do której już odwołuje się Andy ) oraz Dane Polyglot autorstwa Grega Younga.

Jeśli dobrze rozumiem, bieżący przepływ to:

1. Odbieranie i zdarzenie (zakładam przez HTTP?)
2. Poproś o połączenie z puli.
3. Wstaw zdarzenie do bazy danych
4. Zwolnij połączenie z pulą.

Jeśli tak, myślę, że pierwszą zmianą w projekcie byłoby zaprzestanie obsługi połączeń zwrotnych kodu do puli przy każdym zdarzeniu. Zamiast tego utwórz pulę wątków / procesów wstawiania, która jest równa 1 do 1 z liczbą połączeń DB. Każdy z nich utrzyma dedykowane połączenie DB.

Korzystając z pewnego rodzaju współbieżnej kolejki, wątki pobierają wiadomości ze współbieżnej kolejki i wstawiają je. Teoretycznie nigdy nie muszą zwracać połączenia z pulą ani żądać nowego, ale może być konieczne wbudowanie obsługi na wypadek, gdyby połączenie uległo awarii. Najłatwiej może zabić wątek / proces i rozpocząć nowy.

Powinno to skutecznie wyeliminować obciążenie puli połączeń. Będziesz oczywiście musiał wykonać wypychanie co najmniej 1000 / połączeń zdarzeń na sekundę dla każdego połączenia. Możesz wypróbować inną liczbę połączeń, ponieważ 500 połączeń pracujących w tych samych tabelach może spowodować konflikt na DB, ale to zupełnie inne pytanie. Inną rzeczą do rozważenia jest użycie wstawek wsadowych, tj. Każdy wątek wyciąga pewną liczbę wiadomości i wpycha je naraz. Unikaj też wielu połączeń próbujących zaktualizować te same wiersze.

Założenia

Zakładam, że opisywane obciążenie jest stałe, ponieważ jest to trudniejszy scenariusz do rozwiązania.

Zakładam również, że masz jakiś sposób uruchamiania wyzwalanych, długotrwałych obciążeń poza procesem aplikacji sieci Web.

Rozwiązanie

Zakładając, że poprawnie zidentyfikowałeś wąskie gardło - opóźnienie między procesem a bazą danych Postgres - jest to główny problem do rozwiązania. Rozwiązanie musi uwzględniać ograniczenia spójności z innymi klientami, którzy chcą czytać zdarzenia tak szybko, jak to możliwe po ich otrzymaniu.

Aby rozwiązać problem z opóźnieniem, musisz pracować w sposób, który minimalizuje czas oczekiwania na zdarzenie, które ma być przechowywane. Jest to kluczowa rzecz, którą musisz osiągnąć, jeśli nie chcesz lub nie możesz zmienić sprzętu . Biorąc pod uwagę, że korzystasz z usług PaaS i nie masz kontroli nad sprzętem ani siecią, jedynym sposobem zmniejszenia opóźnień na zdarzenie będzie jakiś rodzaj zapisu partii zdarzeń.

Będziesz musiał przechowywać lokalnie kolejkę zdarzeń, która jest okresowo czyszczona i zapisywana do bazy danych, albo po osiągnięciu określonego rozmiaru, albo po upływie określonego czasu. Proces będzie musiał monitorować tę kolejkę, aby uruchomić opróżnianie do sklepu. Powinno być mnóstwo przykładów na temat zarządzania współbieżną kolejką, która jest okresowo opróżniana w wybranym przez Ciebie języku - Oto przykład w języku C # , z okresowego zlewu dozującego popularnej biblioteki rejestrowania Serilog.

Ta odpowiedź SO opisuje najszybszy sposób opróżnienia danych w Postgresie - chociaż wymagałoby to przechowywania wsadowego w kolejce na dysku i prawdopodobnie istnieje problem, który można rozwiązać, gdy dysk zniknie po ponownym uruchomieniu w Heroku.

Przymus

Inna odpowiedź już wspomniała o CQRS i jest to prawidłowe podejście do rozwiązania dla ograniczenia. Chcesz nawodnić modele odczytu podczas przetwarzania każdego zdarzenia - wzorzec Mediator może pomóc w kapsułkowaniu zdarzenia i rozprowadzeniu go do wielu procedur obsługi w toku. Tak więc jeden moduł obsługi może dodać zdarzenie do modelu odczytu, który jest w pamięci, do którego klienci mogą wyszukiwać zapytania, a inny moduł obsługi może być odpowiedzialny za umieszczenie zdarzenia w kolejce pod kątem ewentualnego zapisu grupowego.

Kluczową zaletą CQRS jest rozłączenie koncepcyjnych modeli odczytu i zapisu - co jest fantazyjnym sposobem powiedzenia, że ​​piszesz w jednym modelu, a czytasz w innym całkowicie innym modelu. Aby uzyskać korzyści skalowalności z CQRS, zazwyczaj chcesz mieć pewność, że każdy model jest przechowywany osobno, w sposób optymalny dla wzorców użytkowania. W takim przypadku możemy użyć zagregowanego modelu odczytu - na przykład pamięci podręcznej Redis lub po prostu w pamięci - aby zapewnić, że nasze odczyty są szybkie i spójne, podczas gdy nadal używamy naszej transakcyjnej bazy danych do zapisywania danych.

Zdarzenia przychodzą szybciej niż pula połączeń DB może obsłużyć

Jest to problem, jeśli każdy proces wymaga jednego połączenia z bazą danych. System powinien zostać zaprojektowany tak, abyś miał pulę pracowników, w której każdy pracownik potrzebuje tylko jednego połączenia z bazą danych, a każdy pracownik może przetwarzać wiele zdarzeń.

Kolejka komunikatów może być używana z tym projektem, potrzebujesz producentów komunikatów, którzy wypychają zdarzenia do kolejki komunikatów, a pracownicy (konsumenci) przetwarzają wiadomości z kolejki.

Inni klienci mogą chcieć czytać zdarzenia jednocześnie

To ograniczenie jest możliwe tylko wtedy, gdy zdarzenia przechowywane w bazie danych bez przetwarzania (zdarzenia pierwotne). Jeśli zdarzenia są przetwarzane przed zapisaniem w bazie danych, jedynym sposobem na uzyskanie zdarzeń są dane z bazy danych.

Jeśli klienci chcą po prostu zapytać o nieprzetworzone zdarzenia, sugerowałbym użycie wyszukiwarki takiej jak Elastic Search. Otrzymasz nawet interfejs API zapytania / wyszukiwania za darmo.

Biorąc pod uwagę, że wydaje się, że zapytania dotyczące zdarzeń przed ich zapisaniem w bazie danych są dla Ciebie ważne, proste rozwiązanie, takie jak Elastic Search, powinno działać. Zasadniczo po prostu przechowujesz w nim wszystkie zdarzenia i nie kopiujesz tych samych danych, kopiując je do bazy danych.

Skalowanie Elastyczne wyszukiwanie jest łatwe, ale nawet przy podstawowej konfiguracji jest dość wydajne.

Gdy potrzebujesz przetwarzania, Twój proces może pobrać zdarzenia z ES, przetworzyć je i przechowywać w bazie danych. Nie wiem, jakiego poziomu wydajności potrzebujesz od tego przetwarzania, ale byłoby to całkowicie niezależne od sprawdzania zdarzeń z ES. I tak nie powinieneś mieć problemu z połączeniem, ponieważ możesz mieć stałą liczbę pracowników i każdego z jednym połączeniem z bazą danych.

1k lub 2k zdarzeń (5KB) na sekundę nie jest tak dużo dla bazy danych, jeśli ma odpowiedni schemat i silnik pamięci. Jak sugeruje @eddyce, master z jednym lub więcej slave'ami może oddzielić zapytania o odczyt od wykonywania zapisów. Korzystanie z mniejszej liczby połączeń DB zapewni lepszą ogólną przepustowość.

Inni klienci mogą chcieć czytać zdarzenia jednocześnie

W przypadku tych żądań musieliby również czytać z głównego db, ponieważ opóźnienie replikacji występowałoby w przypadku niewolników odczytu.

Użyłem (Percona) MySQL z silnikiem TokuDB do zapisu bardzo dużych ilości. Istnieje również silnik MyRocks oparty na LSMtrees, który jest dobry do zapisu obciążeń. Zarówno dla tych silników, jak i prawdopodobnie również dla PostgreSQL, istnieją ustawienia izolacji transakcji, a także zachowania synchronizacji zatwierdzania, które mogą znacznie zwiększyć pojemność zapisu. W przeszłości akceptowaliśmy do 1 utraconych danych, które zostały zgłoszone klientowi db jako zatwierdzone. W innych przypadkach były dyski SSD z podtrzymaniem bateryjnym, aby uniknąć utraty.

Amazon RDS Aurora o smaku MySQL ma 6-krotnie wyższą przepustowość zapisu przy zerowej replikacji (podobnie jak niewolnicy współdzielący system plików z masterem). Smak Aurora PostgreSQL ma również inny zaawansowany mechanizm replikacji.

Porzuciłbym heroku razem, to znaczy porzuciłbym podejście scentralizowane: wiele zapisów, które osiągają maksymalne połączenie puli, jest jednym z głównych powodów, dla których klastry db zostały wymyślone, głównie dlatego, że nie ładujesz zapisu db (s) z żądaniami odczytu, które mogą być wykonywane przez inne bazy danych w klastrze, ponadto spróbowałbym z topologią master-slave, ponadto - jak ktoś już wspomniał, posiadanie własnych instalacji db umożliwiłoby dostrojenie całości system, aby upewnić się, że czas propagacji zapytania będzie poprawnie obsługiwany.

Powodzenia