🥇Asynchroniczne programowanie w JavaScript. (Callback, Promise, RxJs)

Cześć wszystkim. Z tej strony Siergiej Omielnicki. Niedawno prowadziłem streaming na temat programowania reaktywnego, gdzie omawiałem asynchroniczność w JavaScript. Dziś chciałbym podsumować ten materiał.

Ale zanim przejdziemy do głównego tematu, musimy zrobić wprowadzenie. Zatem zaczynajmy od definicji: co to jest stos i kolejka?
Stos to zbiór, którego elementy są pobierane według zasady „ostatni wszedł, pierwszy wyszedł” LIFO.
Kolejka to zbiór, którego elementy są pobierane według zasady „pierwszy wszedł, pierwszy wyszedł” FIFO.

Dobrze, kontynuujmy.

JavaScript to język programowania jednowątkowego. Oznacza to, że istnieje tylko jeden wątek wykonania i jeden stos, w którym umieszczane są funkcje w kolejce do wykonania. W związku z tym w danym momencie JavaScript może wykonać tylko jedną operację, a inne operacje będą czekać w kolejce na stosie, aż zostaną wywołane.

Stos wywołań to struktura danych, która, mówiąc w uproszczeniu, zapisuje informacje o miejscu w programie, w którym się znajdujemy. Kiedy przechodzimy do funkcji, umieszczamy jej zapis na górze stosu. Kiedy wracamy z funkcji, zdejmuje się z stosu najwyższy element i wracamy do miejsca, skąd wywołano tę funkcję. To wszystko, co potrafi stos. A teraz bardzo ciekawe pytanie. Jak zatem działa asynchroniczność w JavaScript?

W rzeczywistości oprócz stosu w przeglądarkach istnieje specjalna kolejka dla pracy z tzw. WebAPI. Funkcje z tej kolejki będą wykonywane w kolejności dopiero po całkowitym opróżnieniu stosu. Dopiero wtedy trafiają z kolejki do stosu do wykonania. Jeśli w stosie jest chociaż jeden element, nie mogą się tam dostać. To właśnie dlatego wywołania funkcji z opóźnieniem często są niedokładne, ponieważ funkcja nie może przejść z kolejki do stosu, dopóki ten jest pełen.

Rozważmy następujący przykład i zajmijmy się jego krokowym „wykonaniem”. Również przyjrzyjmy się temu, co się dzieje w systemie.

console.log('Cześć');
setTimeout(function cb1() {
    console.log('cb1');
}, 5000);
console.log('Pa');

1) Na razie nic się nie dzieje. Konsola przeglądarki jest czysta, stos wywołań pusty.

2) Potem polecenie console.log(‘Cześć’) zostaje dodane do stosu wywołań.

3) I zostaje wykonane.

4) Następnie console.log(‘Cześć’) jest usuwane ze stosu wywołań.

5) Teraz przechodzimy do polecenia setTimeout(function cb1() {… }). Zostaje ono dodane do stosu wywołań.

6) Polecenie setTimeout(function cb1() {… }) jest wykonywane. Przeglądarka tworzy timer, który jest częścią Web API. Przeprowadzi on odliczanie czasu.

7) Polecenie setTimeout(function cb1() {… }) zakończyło działanie i zostało usunięte ze stosu wywołań.

8) Polecenie console.log(‘Bye’) zostaje dodane do stosu wywołań.

9) Polecenie console.log(‘Bye’) jest wykonywane.

10) Polecenie console.log(‘Bye’) zostaje usunięte ze stosu wywołań.

11) Po upływie co najmniej 5000 ms., timer kończy działanie i umieszcza callback cb1 w kolejce callbacków.

12) Pętla zdarzeń bierze funkcję cb1 z kolejki callbacków i umieszcza ją w stosie wywołań.

13) Funkcja cb1 jest wykonywana i dodaje console.log(‘cb1’) do stosu wywołań.

14) Polecenie console.log(‘cb1’) jest wykonywane.

15) Polecenie console.log(‘cb1’) zostaje usunięte ze stosu wywołań.

16) Funkcja cb1 zostaje usunięta ze stosu wywołań.

Przyjrzyjmy się przykładzie w akcji:

No i omówiliśmy, jak w JavaScript zrealizowana jest asynchroniczność. Teraz porozmawiajmy krótko o ewolucji kodu asynchronicznego.

Ewolucja kodu asynchronicznego.

a(function (resultsFromA) {
    b(resultsFromA, function (resultsFromB) {
        c(resultsFromB, function (resultsFromC) {
            d(resultsFromC, function (resultsFromD) {
                e(resultsFromD, function (resultsFromE) {
                    f(resultsFromE, function (resultsFromF) {
                        console.log(resultsFromF);
                    })
                })
            })
        })
    })
});

Programowanie asynchroniczne, jakie znamy w JavaScript, może być realizowane tylko za pomocą funkcji. Mogą być one przekazywane jak każda inna zmienna do innych funkcji. Stąd wzięły się callbacki. I jest to fajne, przyjemne i wesołe, dopóki nie zamienia się w smutek, tęsknotę i żal. Dlaczego? Bo wszystko jest proste:

Wraz ze wzrostem złożoności kodu, projekt szybko zamienia się w mało zrozumiałe, wielokrotnie zagnieżdżone bloki — „callback hell”.
Można łatwo przeoczyć obsługę błędów.
Nie można zwracać wyrażeń za pomocą return.

Z pojawieniem się Promise sytuacja stała się trochę lepsza.

new Promise(function(resolve, reject) {
    setTimeout(() => resolve(1), 2000);

}).then((result) => {
    alert(result);
    return result + 2;

}).then((result) => {
    throw new Error('FAILED HERE');
    alert(result);
    return result + 2;

}).then((result) => {
    alert(result);
    return result + 2;

}).catch((e) => {
    console.log('error: ', e);
});

Pojawiły się łańcuchy promes, co poprawiło czytelność kodu.
Pojawiła się osobna metoda do przechwytywania błędów.
Pojawiła się możliwość równoległego wykonywania z użyciem Promise.all.
Zagnieżdżoną asynchroniczność możemy rozwiązać za pomocą async/await

Jednakże obiekt Promise ma swoje ograniczenia. Na przykład, obietnicy nie można anulować bez dodatkowych zabiegów, a co najważniejsze — działa ona tylko z jedną wartością.

No to płynnie przeszliśmy do programowania reaktywnego. Zmęczeni? Na szczęście można zrobić sobie herbatę, przemyśleć to i wrócić do dalszej lektury. A ja kontynuuję.

Programowanie reaktywne — paradigma programowania, która koncentruje się na przepływie danych i propagacji zmian. Przyjrzyjmy się dokładniej, co oznacza przepływ danych.

// Получаем ссылку на элемент
const input = ducument.querySelector('input');

const eventsArray = [];

// Пушим каждое событие в массив eventsArray
input.addEventListener('keyup',
    event => eventsArray.push(event)
);

Wyobraźmy sobie, że mamy pole wprowadzania. Tworzymy tablicę, a dla każdego zdarzenia keyup w polu input zapisujemy to zdarzenie w naszej tablicy. Chciałbym zaznaczyć, że nasza tablica jest posortowana według czasu — tzn. indeks późniejszych zdarzeń jest większy niż indeks wcześniejszych. Taka tablica stanowi uproszczony model przepływu danych, ale to jeszcze nie przepływ. Aby można było śmiało nazwać tę tablicę przepływem, musi ona w jakiś sposób informować subskrybentów o przybyciu nowych danych. W ten sposób doszliśmy do definicji przepływu.

Strumień danych

const { interval1 } = Rx;
const { take } = RxOperators;

interval(1000).pipe(
    take(4)
)

Przepływ — to tablica danych, posortowanych według czasu, która może informować o tym, że dane uległy zmianie. A teraz wyobraźcie sobie, jak wygodnie staje się pisanie kodu, w którym jedno działanie wymaga wywołania kilku zdarzeń w różnych częściach kodu. Po prostu subskrybujemy przepływ, a on sam nas poinformuje, kiedy zajdą zmiany. I to potrafi biblioteka RxJs.

RxJS — to biblioteka do pracy z programami asynchronicznymi i opartymi na zdarzeniach, wykorzystująca obserwowalne sekwencje. Biblioteka oferuje podstawowy typ Observable, kilka typów pomocniczych (Observer, Schedulers, Subjects) oraz operatory pracy z zdarzeniami jako zbiorami (map, filter, reduce, every i podobne z JavaScript Array).

Przyjrzyjmy się głównym pojęciom tej biblioteki.

Observable, Observer, Producer

Observable — pierwszy podstawowy typ, który omówimy. Ta klasa zawiera w sobie główną część implementacji RxJs. Jest powiązana z obserwowalnym przepływem, do którego można subskrybować za pomocą metody subscribe.

W Observable implementowany jest pomocniczy mechanizm do tworzenia aktualizacji, tzw. Obserwator. Źródłem wartości dla Obserwatora jest Producent. Może to być tablica, iterator, web socket, jakieś zdarzenie itd. Można więc powiedzieć, że observable jest pośrednikiem między Producentem a Obserwatorem.

Observable obsługuje trzy rodzaje zdarzeń Obserwatora:

next – nowe dane
error – błąd, jeśli sekwencja zakończyła się z powodu wyjątkowej sytuacji. To zdarzenie również zakłada zakończenie sekwencji.
complete — sygnał o zakończeniu sekwencji. Oznacza to, że nowe dane już nie będą dostępne.

Zobaczmy demo:

Na początku przetworzymy wartości 1, 2, 3, a po 1 sekundzie otrzymamy 4 i zakończymy nasz strumień.

Myśli na głos

I wtedy zrozumiałem, że opowiadanie było ciekawsze niż pisanie o tym. 😀

Subskrypcja

Gdy subskrybujemy strumień, tworzymy nową klasę subskrypcja, która pozwala nam anulować subskrypcję za pomocą metody unsubscribe. Możemy też grupować subskrypcje za pomocą metody add. Oczywiście możemy również rozdzielać strumienie za pomocą usuń. Metody add i remove przyjmują inną subskrypcję jako argument. Warto zauważyć, że gdy się wypisujemy, anulujemy wszystkie subskrypcje podrzędne, jakbyśmy wywoływali metodę unsubscribe także na nich. Przechodzimy dalej.

Rodzaje strumieni

Gorący	Zimny
Producent jest tworzony na zewnątrz observable	Producent jest tworzony wewnątrz observable
Dane przekazywane są w momencie tworzenia observable	Dane są informowane w momencie subskrypcji
Potrzebna jest dodatkowa logika do wypisania się	Strumień kończy się samodzielnie
Używa połączenia jeden-do-wielu	Używa połączenia jeden-do-jednego
Wszystkie subskrypcje mają tę samą wartość	Subskrypcje są niezależne
Dane mogą zostać utracone, jeśli nie ma subskrypcji	Przekazuje wszystkie wartości strumienia dla nowej subskrypcji

Porównując to z analogią, wyobrażam sobie gorący strumień jak film w kinie. W momencie, w którym przyszedłeś, od tego momentu zacząłeś oglądanie. Zimny strumień porównałbym ze złożeniem telefonu do pomocy technicznej. Każdy dzwoniący słucha nagrania automatycznej sekretarki od początku do końca, ale możesz zakończyć połączenie za pomocą unsubscribe.

Chciałbym zauważyć, że istnieją także tzw. ciepłe strumienie (takie określenie spotykałem bardzo rzadko i tylko w zagranicznych społecznościach) – są to strumienie, które przekształcają się z zimnego strumienia w gorący. Pojawia się pytanie – gdzie je wykorzystać? Podam przykład z praktyki.

Pracuję z Angularem. Aktywnie korzysta on z rxjs. Aby uzyskać dane z serwera, oczekuję zimnego strumienia i ten strumień wykorzystuję w szablonie za pomocą asyncPipe. Jeśli użyję tego pipa kilka razy, wracając do definicji zimnego strumienia, każde wywołanie pipe będzie wymagało pobrania danych z serwera, co delikatnie mówiąc jest dziwne. A jeśli przekształcę zimny strumień w ciepły, to zapytanie zostanie wykonane tylko raz.

Ogólnie zrozumienie rodzaju strumieni jest dość trudne dla początkujących, ale ważne.

Operatory

return this.http.get(`${environment.apiUrl}/${this.apiUrl}/trade_companies`)
    .pipe(
        tap(({ data }: TradeCompanyList) => this.companies$$.next(cloneDeep(data))),
        map(({ data }: TradeCompanyList) => data)
    );

Rozszerzenie możliwości pracy ze strumieniami zapewniają operatory. Pomagają one kontrolować zdarzenia zachodzące w Observable. Przeanalizujemy kilka najbardziej popularnych, a bardziej szczegółowe informacje na temat operatorów można znaleźć w linkach w sekcji przydatnych informacji.

Operatory — of

Zacznijmy od pomocniczego operatora of. Tworzy On Observable na podstawie prostej wartości.

Operatory — filter

Operator filtrowania filter, jak można zrozumieć z nazwy, filtruje sygnał strumienia. Jeśli operator zwraca prawdę, to przepuszcza dalej.

Operatory — take

take — Przyjmuje wartość liczby emisji, po której kończy strumień.

Operatory — debounceTime

debounceTime — odrzuca emitowane wartości, które mieszczą się w określonym przedziale czasowym między danymi wyjściowymi — po upływie tego czasu emituje ostatnią wartość.

const { Observable } = Rx;
const { debounceTime, take } = RxOperators;

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Emitujemy wartość co 1000ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);

  // Emitujemy wartość co 1500ms
  setInterval(() => {
    observer.next(i++)
  }, 1500);
}).pipe(
  debounceTime(700),  // Czekamy 700ms na wartość zanim ją przetworzymy
  take(3)
);

Operatory — takeWhile

Emituje wartości, dopóki takeWhile nie zwróci false, po czym wypisuje się ze strumienia.

const { Observable } = Rx;
const { debounceTime, takeWhile } = RxOperators;

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Emit value every 1000ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);
}).pipe(
  takeWhile( producer =>  producer < 5 )
);

Operatory — combineLatest

Operator combineLatest jest podobny do promise.all. Łączy kilka strumieni w jeden. Po tym, jak każdy strumień wyemituje przynajmniej jedną wartość, otrzymujemy ostatnie wartości od każdego w postaci tablicy. Następnie, po każdym emisji z połączonych strumieni, będą przekazywane nowe wartości.

const { combineLatest, Observable } = Rx;
const { take } = RxOperators;

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Emit wartość co 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Emit wartość co 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

combineLatest(observer_1, observer_2).pipe(take(5));

Operatory — zip

Zip czeka na wartość z każdego strumienia i tworzy tablicę na podstawie tych wartości. Jeśli wartość nie dotrze z jakiegokolwiek strumienia, grupa nie zostanie utworzona.

const { zip, Observable } = Rx;
const { take } = RxOperators;

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Emit wartość co 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Emit wartość co 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Emit wartość co 500ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
});

zip(observer_1, observer_2, observer_3).pipe(take(5));

Operatory — forkJoin

forkJoin także łączy strumienie, ale emituje wartość tylko wtedy, gdy wszystkie strumienie zostaną zakończone (complete).

const { forkJoin, Observable } = Rx;
const { take } = RxOperators;

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Emit wartość co 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
}).pipe(take(3));

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Emit wartość co 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
}).pipe(take(5));

const observer_3 = Observable.create((observer) => {
  let i = 1;
  // Emit wartość co 500ms
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
}).pipe(take(4));

forkJoin(observer_1, observer_2, observer_3);

Operatory — map

Operator transformacji map przekształca wartość emisji na nową.

const {  Observable } = Rx;
const { take, map } = RxOperators;

Observable.create((observer) => {
  let i = 1;
  // Emit wartość co 1000ms
  setInterval(() => {
    observer.next(i++);
  }, 1000);
}).pipe(
  map(x => x * 10),
  take(3)
);

Operatory – share, tap

Operator tap – umożliwia wykonywanie efektów ubocznych, to znaczy działań, które nie wpływają na sekwencję.

Operator pomocniczy share potrafi zamienić zimny strumień w gorący.

Z operatorami skończyliśmy. Przejdźmy do Subject.

Myśli na głos

I tu poszedłem na herbatę. Utrudzili mnie te przykłady 😀

Rodzina subject-ów

Rodzina subject-ów jest wyraźnym przykładem gorących strumieni. Te klasy są pewnym hybrydą, które pełnią jednocześnie rolę observable i observer. Ponieważ subject jest gorącym strumieniem, należy się od niego wypisać. Jeśli chodzi o główne metody, to są to:

next – przesyłanie nowych danych do strumienia
error – błąd i zakończenie strumienia
complete – zakończenie strumienia
subscribe – subskrybować strumień
unsubscribe – wypisać się ze strumienia
asObservable – transformujemy w obserwatora
toPromise – przekształca w obietnicę

Wyróżnia się 4-5 typów subject-ów.

Myśli na głos

Na strumieniu mówiłem o 4, a okazało się, że dodali jeszcze jeden. Jak to mówią, całe życie się ucz.

Prosty Subject new Subject()– najprostszy rodzaj subject-ów. Tworzony bez parametrów. Przesyła wartości przychodzące tylko po subskrypcji.

BehaviorSubject new BehaviorSubject(defaultData<T>) – moim zdaniem najpowszechniej stosowany rodzaj subject-ów. Przyjmuje wartość domyślną jako argument. Zawsze przechowuje dane ostatniego emitatu, które przesyła przy subskrypcji. Klasa ta ma również przydatną metodę value, która zwraca bieżącą wartość strumienia.

ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) – Opcjonalnie może przyjąć jako pierwszy argument rozmiar bufora wartości, które będzie przechowywał, a jako drugi czas, w którym potrzebne są zmiany.

AsyncSubject new AsyncSubject() – przy subskrypcji nic się nie dzieje, wartość zostanie zwrócona tylko po complete. Zwróci tylko ostatnią wartość strumienia.

WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig<T> | Observable<T>, destination?: Observer<T>) – O nim dokumentacja milczy, a ja sam widzę go po raz pierwszy. Kto wie, co robi, piszcie, uzupełnimy.

Uff. No i to wszystko, co chciałem dzisiaj opowiedzieć. Mam nadzieję, że te informacje były przydatne. Zbadać listę literatury można w zakładce przydatne informacje.