Używamy cookies, żeby zwiększyć Twoje doświadczenia na stronie

Logowanie/Rejestracja->

Message Queues - system posłańców Imperium

Legionista kończy rejestrację, a serwis wysyła mu list powitalny. Serwer pocztowy akurat nie odpowiada - i rejestracja pada. Człowiek stoi przed bramą odrzucony, choć jego wpis do rejestru powiódł się bez zarzutu. Zawiodła rzecz uboczna, a pociągnęła za sobą rzecz główną.

Rzym rozwiązał to inaczej. Dowódca nie czeka, aż posłaniec wróci z Galii - zostawia list w skrzynce kurierskiej i wraca do swoich zajęć. Posłaniec zabierze go, gdy będzie mógł; jeśli koń okuleje, list poczeka. W systemach nazywamy tę skrzynkę kolejką wiadomości.

Co daje kolejka

Zaleta jest jedna i o nią właśnie chodzi: komunikacja asynchroniczna. Producent - ten, kto wysyła - nie czeka na konsumenta, czyli tego, kto odbiera. Zostawia wiadomość i idzie dalej.

Konsekwencje są trzy. Rejestracja kończy się natychmiast, bo nie jest już związana z pocztą. Awaria konsumenta nie przewraca producenta - wiadomości czekają w kolejce, aż wstanie. A gdy listów przybywa, dokładasz kolejnych konsumentów do tej samej kolejki, nie ruszając producenta.

Producent - zostawienie listu

Aplikacja wysyłająca potrzebuje klienta połączonego z brokerem, czyli z serwerem obsługującym kolejki:

1ClientsModule.register([
2  {
3    name: 'LEGION_SERVICE',
4    transport: Transport.RMQ,
5    options: {
6      urls: ['amqp://localhost:5672'],
7      queue: 'legion_queue',
8    },
9  },
10]);

Transport.RMQ

wskazuje RabbitMQ - najczęściej używanego brokera.

urls

to adres serwera, a

queue

nazwa skrzynki, do której trafią listy. Nazwa

LEGION_SERVICE

posłuży do wstrzyknięcia klienta tam, gdzie będzie potrzebny.

emit kontra send - dwa rodzaje listów

Mając klienta, wysyłamy wiadomość jedną z dwóch metod. Różnica między nimi jest zasadnicza:

1// emit - wyślij i zapomnij, nie czekamy na odpowiedź
2this.client.emit('legion.created', { id: 1, name: 'Legio X' });
3
4// send - wyślij i poczekaj na odpowiedź
5const result = await this.client.send('legion.count', {}).toPromise();

emit()

to fire-and-forget - zostawiasz list i idziesz dalej, nie oczekując odpowiedzi. Tak ogłaszasz fakty: legion powstał, płatność przeszła, użytkownik się zarejestrował. Nadawcy nie interesuje, co odbiorca z tym zrobi.

send()

czeka na odpowiedź - to pytanie, nie ogłoszenie. Używasz go, gdy potrzebujesz wyniku: ile jest legionów, czy ten identyfikator istnieje.

Wybór między nimi decyduje o tym, czy w ogóle korzystasz z zalet kolejki.

send()

przywraca czekanie, a więc i sprzężenie, od którego uciekaliśmy - konsument znów musi działać, żeby producent ruszył dalej. Dlatego do powiadomień, e-maili i logów zawsze

emit()

, @name;

send()

zostaw dla przypadków, gdy naprawdę potrzebujesz odpowiedzi.

Konsument - odebranie listu

Po drugiej stronie kolejki stoi serwis nasłuchujący:

1@Controller()
2export class LegionConsumer {
3  @EventPattern('legion.created')
4  async handleLegionCreated(@Payload() data: LegionCreatedDto) {
5    await this.mailService.sendWelcome(data);
6  }
7}

@EventPattern('legion.created')

mówi: ta metoda obsługuje wiadomości o takiej nazwie. Nazwa jest umową między producentem a konsumentem - musi zgadzać się co do znaku z tą podaną w

emit()

@Payload()

wyciąga treść wiadomości, czyli obiekt przekazany jako drugi argument.

Zauważ, że konsument wygląda jak zwykły kontroler - bo działa podobnie. Różnica polega na tym, skąd przychodzi żądanie: nie z sieci HTTP, tylko z kolejki.

Potwierdzenie - kwit odbioru

Broker musi wiedzieć, czy list dotarł i został przetworzony. Bez tej informacji nie może go usunąć - a gdyby usunął od razu po wysłaniu, awaria konsumenta w połowie pracy oznaczałaby bezpowrotną stratę wiadomości.

Stąd potwierdzenie, czyli acknowledgement:

1@EventPattern('legion.created')
2async handleLegionCreated(@Payload() data: LegionCreatedDto, @Ctx() context: RmqContext) {
3  const channel = context.getChannelRef();
4  const originalMsg = context.getMessage();
5
6  try {
7    await this.mailService.sendWelcome(data);
8
9    channel.ack(originalMsg);
10  } catch (error) {
11    channel.nack(originalMsg, false, true);
12  }
13}

@Ctx()

daje dostęp do kontekstu RabbitMQ.

getChannelRef()

zwraca kanał komunikacji z brokerem, a

getMessage()

- surową wiadomość, której dotyczy potwierdzenie.

Sedno jest w bloku

try

. channel.ack(originalMsg)
potwierdza poprawne przetworzenie - dopiero wtedy broker usuwa wiadomość z kolejki. Zwróć uwagę na miejsce tego wywołania: po wykonaniu pracy, nie przed. Potwierdzenie na początku metody znaczyłoby "odebrałem", a nie "przetworzyłem", i awaria wysyłki e-maila skończyłaby się cichą utratą listu.

catch

stoi

nack

- odmowa potwierdzenia. Ostatni argument

true

każe brokerowi wstawić wiadomość z powrotem do kolejki, żeby spróbować później. To jest właśnie ta odporność, dla której kolejki wprowadziliśmy: nieudana próba nie niszczy wiadomości.

Podsumowanie

Skrzynka kurierska działa, listy dochodzą:

kolejka wiadomości daje komunikację asynchroniczną - producent nie czeka na konsumenta,
dzięki temu awaria konsumenta nie przewraca producenta, a wiadomości czekają, aż odbiorca wstanie,
klienta konfigurujemy przez
```
ClientsModule.register
```
z
```
Transport.RMQ
```
, adresem brokera i nazwą kolejki,
emit()
to fire-and-forget - ogłaszasz fakt i nie czekasz na odpowiedź; send()
czeka na odpowiedź, więc przywraca sprzężenie,
do powiadomień i e-maili używaj
```
emit()
```
;
```
send()
```
tylko wtedy, gdy naprawdę potrzebujesz wyniku,
konsument nasłuchuje przez
```
@EventPattern('nazwa')
```
, a treść wyciąga
```
@Payload()
```
; nazwa musi zgadzać się z tą z
```
emit()
```
,
channel.ack(originalMsg)
potwierdza przetworzenie i dopiero wtedy broker usuwa wiadomość z kolejki,
potwierdzaj po wykonaniu pracy, nie przed - inaczej awaria oznacza cichą utratę wiadomości,
```
nack
```
z ostatnim argumentem
```
true
```
wraca wiadomość do kolejki do ponownej próby,
pełny przepływ: producent wysyła przez
```
emit
```
→ wiadomość czeka w kolejce → konsument odbiera przez
```
@EventPattern
```
→ potwierdza przez
```
channel.ack
```
.

To ostatnia lekcja tego modułu. Umiesz już sprawdzić każdą część osobno, ich współpracę, całą drogę żądania, zmierzyć pokrycie i wytrzymałość, a teraz także rozdzielić serwisy tak, by awaria jednego nie pociągała reszty. A na razie zapamiętaj: kolejka to skrzynka kurierska - dowódca zostawia list i wraca do swoich spraw, a potwierdzenie odbioru przychodzi dopiero wtedy, gdy posłaniec naprawdę dotarł.

Przejdź do CodeWorlds→