🥇Programimi asinkron në JavaScript. ( Callback, Promise, RxJs )

Përshëndetje të gjithëve. Këtu është Omelnytskiy Serhiy. Para pak kohe mbajta një transmetim për programimin reaktiv, ku flas për asinkronitetin në JavaScript. Sot do doja të përmbledh këtë material.

Por para se të fillojmë me materialin kryesor, na nevojitet një hyrje. Le të fillojmë me definita: çfarë është staku dhe radhitja?
Staku — është një koleksion, elementet e të cilit merret sipas parimit "i fundit i hyrë, i pari i dalë" LIFO
Radhitja — është një koleksion, elementet e të cilit merret sipas parimit "i pari i hyrë, i pari i dalë" FIFO

Mirë, të vazhdojmë.

JavaScript është një gjuhë programimi njëpërmasore. Kjo do të thotë se ka vetëm një rrjedhë ekzekutimi dhe një stok, në të cilin vendosen funksionet në radhë për ekzekutim. Si pasojë, në një moment të vetëm JavaScript mund të ekzekutojë vetëm një operacion, operacione të tjera do të presin në radhë derisa të thirren.

Staku i thirrjeve — është një strukturë të dhënash që, në terma të thjeshtë, regjistron informacionin mbi vendin në program ku ndodhemi. Nëse kalojmë në një funksion, ne vendosim një regjistrim për të në majë të stack-ut. Kur kthehemi nga funksioni, ne nxjerrim elementin më të lartë nga stack-u dhe përfundojmë atje ku e kemi thirrur këtë funksion. Kjo është e gjitha që mund të bëjë stack-u. Tani një pyetje mjaft interesante. Si funksionon atëherë asinkroniciteti në JavaScript?

Në të vërtetë, përveç stack-ut, në shfletues ka një radhë të veçantë për të punuar me atë që quhet WebAPI. Funksionet nga kjo radhë do të ekzekutohen në rendësi vetëm pasi stack-u të jetë plotësisht i zbrazët. Vetëm pas kësaj ato vendosen nga radhë në stack për t'u ekzekutuar. Nëse në stack aktualisht ndodhet të paktën një element, ato nuk mund të hyjnë në stack. Pikërisht për këtë arsye thirrjet e funksioneve me timeout shpesh nuk janë të sakta në kohë, sepse funksioni nuk mund të hyjë nga radhë në stack derisa ai të jetë i plotë.

Le të shqyrtojmë shembullin e mëposhtëm dhe të merremi me "ekzekutimin" e tij hap pas hapi. Po ashtu, do të shikojmë se çfarë ndodh në sistem gjatë kësaj procedure.

console.log('Hi);
setTimeout(function cb1() {
    console.log('cb1');
}, 5000);
console.log('Bye');

1) Deri asnjë gjë. Konsola e shfletuesit është e pastër, staku i thirrjeve është i zbrazët.

2) Më pas, komanda console.log(‘Hi’) shtohet në stakun e thirrjeve.

3) Dhe ajo ekzekutohet.

4) Pjesa console.log(‘Hi’) hiqet nga staku i thirrjeve.

5) Tani kalojmë te komanda setTimeout(function cb1() {… }). Ajo shtohet në stakun e thirrjeve.

6) Komanda setTimeout(function cb1() {… }) ekzekutohet. Shfletuesi krijon një timer që është pjesë e Web API. Ai do të bëjë numërimin mbrapsht.

7) Komanda setTimeout(function cb1() {… }) përfundon dhe hiqet nga staku i thirrjeve.

8) Komanda console.log(‘Bye’) shtohet në stakun e thirrjeve.

9) Komanda console.log(‘Bye’) ekzekutohet.

10) Komanda console.log(‘Bye’) hiqet nga staku i thirrjeve.

11) Pasi të kalojnë, të paktën, 5000 ms., timer-i përfundojnë dhe vendos kolb-bekimin cb1 në radhën e kolb-bekimeve.

12) Cikli i ngjarjeve merr funksionin cb1 nga radhë e kolb-bekimeve dhe e vendos atë në stakun e thirrjeve.

13) Funksioni cb1 ekzekutohet dhe shton console.log(‘cb1’) në stakun e thirrjeve.

14) Komanda console.log(‘cb1’) ekzekutohet.

15) Komanda console.log(‘cb1’) hiqet nga staku i thirrjeve.

16) Funksioni cb1 hiqet nga staku i thirrjeve.

Shikojmë një shembull në dinamikë:

Tani që kemi shqyrtuar se si është implementuar asinkronia në JavaScript, le të flasim shkurtimisht mbi evolucionin e kodit asinkron.

Evolucioni i kodit asinkron.

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);
                    })
                })
            })
        })
    })
});

Programimi asinkron, siç e dimë ne JavaScript, mund të implementohet vetëm me funksione. Ato mund të kalohen si çdo variabël tjetër në funksione të tjera. Kështu lindën callback-ët. Dhe kjo është argëtuese dhe gëzuese derisa të shndërrohet në trishtim, mërzi dhe brengë. Pse? Sepse është shumë e thjeshtë:

Me rritjen e kompleksitetit të kodit, projekti shpejt shndërrohet në blloqe të thelluara të paqartë - "callback hell".
Përpunimi i gabimeve lehtësisht mund të humbasë.
Nuk është e mundur të kthehen shprehje me return.

Me ardhjen e Promise, situata u bë pak më e mirë.

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);
});

Me shfaqen zinxhirë promisesh, që e përmirësojnë lexueshmërinë e kodit
Shfaqet një metodë e veçantë për kapjen e gabimeve
Shfaqet mundësia e ekzekutimit paralel me ndihmën e Promise.all
Asinhronësinë e brendshme mund ta zgjidhim me async/await

Por promesa ka kufizimet e saj. Për shembull, një promise, pa ndonjë marrëveshje të veçantë, nuk mund të anulohet, dhe më e rëndësishmja — punon vetëm me një vlerë.

Tani ne kemi arritur smooth në programimin reaktiv. A jeni të lodhur? Mirë, mund të shkoni të bëni një çaj, të mendoni dhe të ktheheni për të lexuar më tej. Unë do të vazhdoj.

Programimi reaktiv — një paradigme programimi që është e orientuar nga rrjedhat e të dhënave dhe shpërndarjen e ndryshimeve. Le të shqyrtojmë më në detaje se çfarë është rrjedha e të dhënave.

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

const eventsArray = [];

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

Imagjinoni se kemi një fushë hyrëse. Ne krijojmë një masiv dhe për çdo ngjarje keyup të inputit do të ruajmë ngjarjen në masivin tonë. Dëshirojmë të theksojmë se masivi ynë është renditur sipas kohës, pra indeksi i ngjarjeve më të vona është më i madh se indeksi i atyre më të hershme. Ky masiv përfaqëson një model të thjeshtuar të rrjedhjes së të dhënave, por nuk është ende një rrjedhë. Për ta quajtur këtë masiv me siguri një rrjedhë, ai duhet të dijë se si t'i informojë abonuesit për të dhënat e reja që janë marrë. Kështu arritëm në përcaktimin e rrjedhës.

Shkalla e të dhënave

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

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

Rrjedhaështë një masiv të dhënash të renditura sipas kohës, i cili mund t'i njoftojë se të dhënat kanë ndryshuar. Tani imagjinoni sa e lehtë është të shkruani kod, ku për një veprim duhet të thërrisni disa ngjarje në pjesë të ndryshme të kodit. Ne thjesht bëjmë një abonim në rrjedhë dhe ajo na informon vetë kur ndodhin ndryshimet. Këtë e bën biblioteka RxJs.

RxJS — është një bibliotekë për të punuar me programe asinkrone dhe të bazuara në ngjarje duke përdorur sekuenca të vëzhguara. Biblioteca ofron tipin kryesor Observable, disa lloje ndihmëse (Observer, Schedulers, Subjects) dhe operatorë për të punuar me ngjarjet si me koleksionet (map, filter, reduce, every dhe të ngjashme nga JavaScript Array).

Le të shqyrtojmë konceptet kryesore të kësaj biblioteke.

Observable, Observer, Producer

Observable është tipi i parë bazë që do të shqyrtojmë. Ky klas përmban pjesën kryesore të implementimit të RxJs. Ai është i lidhur me fluxin e vëzhguar, në të cilin mund të abonoheni përmes metodës subscribe.

Në Observable implementohet një mekanizëm ndihmës për krijimin e azhurnimeve, e ashtuquajtura Observer. Burimi i vlerave për Observer quhet Prodhues. Kjo mund të jetë një array, një iterues, web socket, një ngjarje etj. Pra, mund të thuhet se observable është një lidhës midis Producer dhe Observer.

Observable përpunon tri lloje ngjarjesh për Observer:

next – të dhëna të reja
error – një gabim, nëse sekuenca përfundoi për shkak të një situate përjashtimi. Ky ngjarje gjithashtu presupozon përfundimin e sekuencës.
komplet - sinjali i përfundimit të sekuencës. Kjo do të thotë se nuk do të ketë më të dhëna të reja

Të shohim një demo:

Në fillim do të përpunojmë vlerat 1, 2, 3, dhe pas 1 sek. do të marrim 4 dhe do të përfundojmë stream-in tonë.

Të mendojmë në zë

Dhe këtu e kuptova se ishte më interesant të tregoja se sa të shkruaja për këtë. 😀

Abonim

Kur ne bëjmë një abonim në stream, krijojmë një klasë të re subscription, e cila na jep mundësinë të anulojmë abonimin me metodën unsubscribe. Gjithashtu, mund të grumbullojmë abonimet me metodën shto. Dhe është logjike që mund të shkëpusim stream-et me remove. Metodat add dhe remove pranojnë një abonim të tjetër si argument. Dëshiroj të theksoj se kur ne bëjmë anullimin, ne anulojmë të gjitha abonimet fëmijë siç do të thoshim se ato do të thërrisnin metodën unsubscribe. Të vazhdojmë më tej.

Llojet e stream-eve

Nxehtë	Ftohtë
Prodhuese krijohet jashtë observable	Prodhuese krijohet brenda observable
Të dhënat transmetohen në momentin e krijimit të observable	Të dhënat komunikohen në momentin e abonimit
Nevojitet logjikë shtesë për anulimin	Stream-i përfundon vetë
Përdor lidhjen një-në-shumë	Përdor lidhjen një-në-një
Të gjitha abonimet kanë një vlerë të vetme	Abonimet janë të pavarura
Të dhënat mund të humbasin nëse nuk ka subscription	Rindez të gjitha vlerat e rrjedhës për një abonim të ri

Nëse do të bënim një krahasim, do ta përshkruaja rrjedhën e ngrohtë si një film në kinema. Në momentin kur ke hyrë, aty e ke filluar shikimin. Rrjedhën e ftohtë do ta krahasoja me një telefonatë në mbështetje teknike. Çdo person që telefonon do të dëgjojë regjistrimin e automatikës nga fillimi deri në fund, por ti mund të ndërpresësh telefonatën duke u unsubscribe.

Dëshiroj të theksoj se ekzistojnë edhe rrjedha të ashtuquajtura të ngrohta (këtë definicion e kam hasur jashtëzakonisht rrallë dhe vetëm në komunitetet ndërkombëtare) – kjo është një rrjedhë që transformohet nga e ftohtë në të ngrohtë. Lind pyetja – ku të përdoret? Do të jap një shembull nga praktika.

Unë punoj me Angular. Ai përdor aktivisht rxjs. Për të marrë të dhëna nga serveri, unë pres një rrjedhë të ftohtë dhe këtë rrjedhë e përdor në shabllon përmes asyncPipe. Nëse e përdor këtë pipë disa herë, duke u kthyer te definicioni i rrjedhës së ftohtë, çdo pipë do të kërkojë të dhëna nga serveri, që është, për ta thënë butë, e çuditshme. Ndërsa nëse e konvertoj rrjedhën e ftohtë në një të ngrohtë, kërkesa do të ndodhë vetëm një herë.

Në të vërtetë, kuptimi i llojeve të rrjedhave është mjaft i komplikuar për fillestarët, por është i rëndësishëm.

Operators

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

Opertorët na ofrojnë mundësi të zgjeruara për të punuar me rrjedhat. Ata ndihmojnë në kontrollin e ngjarjeve që ndodhin në Observable. Do të shqyrtojmë disa nga më të njohurit, ndërsa për më shumë informacion rreth operatorëve mund të konsultoheni me lidhjet në informacionin e dobishëm.

Operators — of

Le të fillojmë me operatorin ndihmës of. Ai krijon një Observable në bazë të një vlerë të thjeshtë.

Operators — filter

Operatori i filtrimit filter, siç mund të kuptohet nga emri, filtrojnë sinjalin e rrjedhës. Nëse operatori kthen një vërtetësi, atëherë e kalon më tej.

Operatorët — merr

merr — Pranon vlerën e numrit të emetimeve pas të cilave mbyll rrjedhën.

Operatorët — debounceTime

debounceTime — eliminon vlerat e emetuara që bien brenda një periudhe kohe të caktuar midis të dhënave të daljes — pas kalimit të intervalit kohor emeton vlerën e fundit.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Emeton vlerën çdo 1000ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);

 // Emeton vlerën çdo 1500ms
  setInterval(() => {
    observer.next(i++)
  }, 1500);
}).pipe(
  debounceTime(700),  // Prit 700ms për vlerën para se ta処ie
  take(3)
);

Operatorët — takeWhile

Emeton vlera derisa takeWhile të kthejë false, pas së cilës do të çregjistrohet nga rrjedha.

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

Observable.create((observer) => {
  let i = 1;
  observer.next(i++);
  // Emeton vlerën çdo 1000ms
  setInterval(() => {
    observer.next(i++)
  }, 1000);
}).pipe(
  takeWhile( prodhues =>  prodhues < 5 )
);

Operatorët — combineLatest

Operatori i kombinuar combineLatest është disi i ngjashëm me promise.all. Ai bashkon disa rrjedha në një të vetëm. Pasi çdo rrjedhë të emitojë të paktën një herë, ne marrim vlerat e fundit nga çdo njëri si një array. Pas çdo emiti nga rrjedhat e kombinuara, ai do të kthejë vlera të reja.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  // Emiton një vlerë çdo 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  // Emiton një vlerë çdo 750ms
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

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

Operatorët — zip

Zip — pret një vlerë nga secila rrjedhë dhe formon një array në bazë të këtyre vlerave. Nëse nuk vjen asnjë vlerë nga ndonjë rrjedhë, grupi nuk do të formohet.

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  \/\/ Emitted value every 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
});

const observer_2 = Observable.create((observer) => {
  let i = 1;
  \/\/ Emitted value every 750
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
});

const observer_3 = Observable.create((observer) => {
  let i = 1;
  \/\/ Emitted value every 500
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
});

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

Operators — forkJoin

forkJoin gjithashtu bashkon rrjedhat, por ai e lëshon vlerën vetëm kur të gjitha rrjedhat të përfundojnë (complete).

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

const observer_1 = Observable.create((observer) => {
  let i = 1;
  \/\/ Emitted value every 1000ms
  setInterval(() => {
    observer.next('a: ' + i++);
  }, 1000);
}).pipe(take(3));

const observer_2 = Observable.create((observer) => {
  let i = 1;
  \/\/ Emitted value every 750
  setInterval(() => {
    observer.next('b: ' + i++);
  }, 750);
}).pipe(take(5));

const observer_3 = Observable.create((observer) => {
  let i = 1;
  \/\/ Emitted value every 500
  setInterval(() => {
    observer.next('c: ' + i++);
  }, 500);
}).pipe(take(4));

forkJoin(observer_1, observer_2, observer_3);

Operators — map

Operatori i transformimit map e ndryshon vlerën e emetimit në një të re.

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

Observable.create((observer) => {
  let i = 1;
  // Çlirojnë vlerën çdo 1000ms
  setInterval(() => {
    observer.next(i++);
  }, 1000);
}).pipe(
  map(x => x * 10),
  take(3)
);

Operatorët – share, tap

Operatori tap lejon të bëhen efekte anësore, domethënë ndonjë veprim që nuk ka ndikim në rendin.

Operatori utilitar share është në gjendje të bëjë nga një rrjedhë të ftohtë një rrjedhë të ngrohtë.

Me operatorët përfunduam. Tani do të kalojmë te Subject.

Të mendojmë në zë

Dhe këtu shkova të pi çaj. Më lodhën këto shembuj 😀

Familja e subject-ëve

Familja e subject-ëve është një shembull i shkëlqyer i rrjedhave të ngrohta. Këto klasa janë një hibrid, që veprojnë njëherazi si observable dhe observer. Duke qenë se subject është një rrjedhë e ngrohtë, është e nevojshme të mbahen në mend se duhet të shkëputeni. Në lidhje me metodat kryesore, ato janë:

next – dërgimi i të dhënave të reja në rrjedhë
error – gabim dhe përfundimi i rrjedhës
complete – përfundimi i rrjedhës
subscribe – abonohem në rrjedhë
unsubscribe – shkëputem nga rrjedha
asObservable – e transformojmë në një vëzhgues
toPromise – e transformon në një premtim

Dallohet 4-5 lloje subject-ësh.

Të mendojmë në zë

Në stream thashë 4, por doli se ata kanë shtuar një tjetër. Siç thonë, jeton dhe mëso!

Subject i thjeshtë new Subject()– lloji më i thjeshtë i subject-eve. Krijohet pa parametra. Transmeton vlerat që vijnë vetëm pas abonimit.

BehaviorSubject new BehaviorSubject( defaultData ) – sipas mendimit tim, lloji më i zakonshëm i subject-eve. Merr një vlerë fillestare. Gjithmonë ruan të dhënat e lëshimit të fundit, të cilat i transmeton gjatë abonimit. Ky klas ka gjithashtu një metodë të dobishme value, e cila kthen vlerën aktuale të rrjedhës.

ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) — Mund të marrë si argument të parë opsional madhësinë e tamponit të vlerave që do të mbajë, dhe si të dytë kohën gjatë së cilës na nevojiten ndryshimet.

AsyncSubject new AsyncSubject() — gjatë abonimit nuk ndodh asgjë, dhe vlera do të kthehet vetëm kur të përfundojë. Do të kthehet vetëm vlera e fundit e rrjedhës.

WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig | Observable, destination?: Observer) — Dokumentacioni nuk thotë asgjë për të, dhe unë e shoh për herë të parë. Kush e di se çfarë bën, ju lutem shkruani, do ta plotësojmë.

Ah, kështu që ne kemi shqyrtuar gjithçka që doja të ndaja sot. Shpresoj që kjo informacion të ketë qenë e dobishme. Mund të shqyrtoni listën e literaturës në seksionin e informacionit të dobishëm.