Olá a todos. Aqui é o Sergey Omelnytsky. Não faz muito tempo, fiz uma transmissão ao vivo sobre programação reativa, onde falei sobre assincronidade em JavaScript. Hoje, gostaria de resumir esse material.
Mas antes de começarmos o conteúdo principal, precisamos fazer uma introdução. Então, vamos começar com definições: o que são pilha e fila?
Pilha — é uma coleção cujos elementos são obtidos segundo o princípio "último a entrar, primeiro a sair" (LIFO).
Fila — é uma coleção cujos elementos são obtidos segundo o princípio "primeiro a entrar, primeiro a sair" (FIFO).
Ok, vamos continuar.
JavaScript é uma linguagem de programação de thread única. Isso significa que há apenas uma thread de execução e uma pilha, na qual as funções são colocadas em espera para serem executadas. Portanto, em um determinado momento, JavaScript só pode executar uma operação; outras operações ficarão aguardando sua vez na pilha até serem chamadas.
Pilha de chamadas — é uma estrutura de dados que, de forma simplificada, grava informações sobre onde estamos no programa. Se entramos em uma função, colocamos uma entrada sobre ela na parte superior da pilha. Quando retornamos de uma função, removemos o elemento do topo da pilha e voltamos ao ponto de onde chamamos essa função. Isso é tudo o que a pilha faz. Agora, surge uma pergunta muito interessante. Como, então, a assincronidade funciona em JavaScript?
Na verdade, além da pilha, os navegadores têm uma fila especial para trabalhar com o chamado WebAPI. As funções dessa fila serão executadas em ordem apenas depois que a pilha estiver completamente limpa. Somente após isso, elas são transferidas da fila para a pilha para execução. Se houver pelo menos um elemento na pilha, elas não poderão entrar na pilha. É exatamente por isso que a chamada de funções de timeout muitas vezes não é precisa em termos de tempo, já que a função não pode ir da fila para a pilha enquanto esta estiver cheia.
Vamos considerar o seguinte exemplo e analisá-lo passo a passo. Também veremos o que está acontecendo no sistema durante isso.
console.log('Hi);
setTimeout(function cb1() {
console.log('cb1');
}, 5000);
console.log('Bye');
1) Enquanto nada acontece. O console do navegador está limpo, a pilha de chamadas está vazia.
2) Depois, o comando console.log('Hi') é adicionado à pilha de chamadas.
3) E ela é executada
4) Em seguida, console.log('Hi') é removido da pilha de chamadas.
5) Agora passamos para o comando setTimeout(function cb1() {… }). Ele é adicionado à pilha de chamadas.
6) O comando setTimeout(function cb1() {… }) é executado. O navegador cria um temporizador, que é parte da API Web. Ele começará a contagem regressiva.
7) O comando setTimeout(function cb1() {… }) terminou e é removido da pilha de chamadas.
8) O comando console.log('Bye') é adicionado à pilha de chamadas.
9) O comando console.log('Bye') é executado.
10) O comando console.log('Bye') é removido da pilha de chamadas.
11) Depois que pelo menos 5000 ms passarem, o temporizador termina e coloca o callback cb1 na fila de callbacks.
12) O loop de eventos pega a função cb1 da fila de callbacks e a coloca na pilha de chamadas.
13) A função cb1 é executada e adiciona console.log('cb1') à pilha de chamadas.
14) O comando console.log('cb1') é executado.
15) O comando console.log('cb1') é removido da pilha de chamadas.
16) A função cb1 é removida da pilha de chamadas.
Vamos olhar para um exemplo em ação:
E assim, examinamos como a assíncronicidade é implementada em JavaScript. Agora, vamos falar brevemente sobre a evolução do código assíncrono.
Evolução do código assíncrono.
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);
})
})
})
})
})
});A programação assíncrona, como conhecemos em JavaScript, pode ser implementada apenas com funções. Elas podem ser passadas como qualquer outra variável para outras funções. Assim, nasceram os callbacks. E isso é legal, divertido e emocionante, até que se transforma em tristeza e desespero. Por quê? É simples:
- Com o aumento da complexidade do código, o projeto rapidamente se transforma em blocos aninhados difíceis de entender — o 'callback hell'.
- O tratamento de erros pode ser facilmente negligenciado.
- Não se pode devolver expressões com return.
Com o surgimento das Promises, a situação melhorou um pouco.
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);
});- As cadeias de promessas surgiram, melhorando a legibilidade do código.
- Um método separado para captura de erros foi introduzido.
- Agora é possível a execução paralela com Promise.all
- Podemos resolver a aninhamento assíncrono com async/await
Mas a promessa tem suas limitações. Por exemplo, uma promessa não pode ser cancelada sem muitos truques, e o mais importante — ela trabalha com um único valor.
Assim, chegamos suavemente à programação reativa. Está cansado? Bem, você pode ir fazer um chá, pensar um pouco e voltar a ler. Eu continuarei.
Programação reativa — é uma paradigma de programação focado em fluxos de dados e propagação de mudanças. Vamos analisar mais detalhadamente o que é um fluxo de dados.
// Получаем ссылку на элемент
const input = ducument.querySelector('input');
const eventsArray = [];
// Пушим каждое событие в массив eventsArray
input.addEventListener('keyup',
event => eventsArray.push(event)
);Vamos imaginar que temos um campo de entrada. Criamos um array e, a cada evento keyup do input, salvamos o evento neste array. Observamos que nosso array está ordenado por tempo, ou seja, o índice dos eventos mais recentes é maior que o índice dos eventos mais antigos. Esse array representa um modelo simplificado de um fluxo de dados, mas ainda não é um fluxo. Para que esse array possa ser considerado um fluxo, ele deve conseguir de alguma forma informar aos assinantes que novos dados foram recebidos. Assim, chegamos à definição de um fluxo.
Fluxo de dados
const { interval1 } = Rx;
const { take } = RxOperators;
interval(1000).pipe(
take(4)
)
Fluxo — é um array de dados ordenado por tempo que pode comunicar quando os dados foram alterados. Agora imagine como se torna conveniente escrever código em que uma ação requer a chamada de vários eventos em diferentes partes do código. Nós apenas nos inscrevemos no fluxo e ele nos avisa quando ocorrem mudanças. E é isso que a biblioteca RxJs pode fazer.
RxJS — é uma biblioteca para trabalhar com programas assíncronos e baseados em eventos usando sequências observáveis. A biblioteca fornece o tipo principal Observable, vários tipos auxiliares (Observer, Schedulers, Subjects) e operadores para trabalhar com eventos como coleções (map, filter, reduce, every e semelhantes do Array JavaScript).
Vamos entender os conceitos básicos desta biblioteca.
Observable, Observer, Producer
Observable — é o primeiro tipo básico que examinaremos. Esta classe contém a parte principal da implementação do RxJs. Ela está relacionada ao fluxo observável, no qual você pode se inscrever usando o método subscribe.
No Observable implementa um mecanismo auxiliar para gerar atualizações, conhecido como Observer. A fonte de valores para o Observer é chamada de Producer. Isso pode ser um array, um iterador, um web socket, um evento, etc. Portanto, podemos dizer que o observable é um condutor entre Producer e Observer.
O Observable processa três tipos de eventos do Observer:
- next – novos dados
- error – um erro, se a sequência for concluída devido a uma situação excepcional. Este evento também implica a conclusão da sequência.
- complete — um sinal de que a sequência foi concluída. Isso significa que não haverá mais novos dados
Vamos ver uma demonstração:
No começo, processaremos os valores 1, 2, 3, e após 1 segundo, receberemos 4 e finalizaremos nosso fluxo.
Pensando em voz alta
E aqui percebi que era mais interessante contar do que escrever sobre isso. 😀
Subscription
Quando fazemos uma assinatura no fluxo, criamos uma nova classe subscription, que nos permite cancelar a assinatura usando o método unsubscribe. Também podemos agrupar assinaturas usando o método add. E é lógico que podemos desagrupar fluxos usando remover. Os métodos add e remove aceitam outra assinatura como entrada. Vale ressaltar que, quando cancelamos a assinatura, estamos cancelando todas as assinaturas filhas como se tivéssemos chamado o método unsubscribe nelas. Vamos em frente.
Tipos de fluxos
| HOT | COLD |
|---|---|
| O Producer é criado fora do observable | O Producer é criado dentro do observable |
| Os dados são transmitidos no momento da criação do observable | Os dados são comunicados no momento da assinatura |
| É necessária lógica adicional para o cancelamento da assinatura | O fluxo se encerra sozinho |
| Usa uma relação um-para-muitos | Usa uma relação um-para-um |
| Todas as assinaturas têm um valor único | As assinaturas são independentes |
| Os dados podem ser perdidos se não houver assinatura | Reemiti todos os valores do fluxo para uma nova assinatura |
Se eu fizer uma analogia, eu representaria um fluxo quente como um filme no cinema. No momento em que você chega, começa a assistir a partir daquele ponto. Um fluxo frio eu compararia a uma chamada para a assistência técnica. Qualquer pessoa que ligar ouve a gravação do correio de voz do começo ao fim, mas você pode desligar com o método unsubscribe.
Gostaria de observar que existem também os chamados fluxos quentes (essa definição que encontrei raramente e apenas em comunidades estrangeiras) — este é um fluxo que se transforma de um fluxo frio em um quente. Surge a pergunta — onde usar isso? Vou dar um exemplo da prática.
Eu trabalho com Angular. Ele utiliza ativamente o RxJS. Para obter dados no servidor, espero um fluxo frio e utilizo esse fluxo no template com o asyncPipe. Se eu usar esse pipe várias vezes, voltando à definição de fluxo frio, cada pipe irá solicitar dados do servidor, o que, para dizer o mínimo, é estranho. E se eu transformar o fluxo frio em quente, a solicitação ocorrerá uma única vez.
Na verdade, a compreensão dos tipos de fluxos é bastante complexa para iniciantes, mas importante.
Operadores
return this.http.get(`${environment.apiUrl}/${this.apiUrl}/trade_companies`)
.pipe(
tap(({ data }: TradeCompanyList) => this.companies$$.next(cloneDeep(data))),
map(({ data }: TradeCompanyList) => data)
);Os operadores nos fornecem a capacidade de trabalhar com fluxos. Eles ajudam a controlar os eventos que ocorrem em um Observable. Vamos considerar alguns dos mais populares; para mais detalhes sobre os operadores, você pode consultar os links nas informações úteis.
Operadores — of
Vamos começar com o operador auxiliar of. Ele cria um Observable com base em um valor simples.
Operadores — filter
O operador de filtragem filter, como o próprio nome sugere, filtra o sinal do fluxo. Se o operador retornar verdadeiro, ele passa adiante.
Operadores — take
take — Aceita um número de emissões, após o qual o fluxo é encerrado.
Operadores — debounceTime
debounceTime — descarta os valores emitidos que ocorrem dentro de um intervalo de tempo especificado entre as saídas — após o intervalo de tempo, emite o último valor.
const { Observable } = Rx;
const { debounceTime, take } = RxOperators;
Observable.create((observer) => {
let i = 1;
observer.next(i++);
// Emite um valor a cada 1000ms
setInterval(() => {
observer.next(i++)
}, 1000);
// Emite um valor a cada 1500ms
setInterval(() => {
observer.next(i++)
}, 1500);
}).pipe(
debounceTime(700), // Espera 700ms pelos valores antes de processá-los
take(3)
); 
Operadores — takeWhile
Emite valores enquanto takeWhile não retornar falso, após o que se desinscreve do fluxo.
const { Observable } = Rx;
const { debounceTime, takeWhile } = RxOperators;
Observable.create((observer) => {
let i = 1;
observer.next(i++);
// Emite um valor a cada 1000ms
setInterval(() => {
observer.next(i++)
}, 1000);
}).pipe(
takeWhile(producer => producer < 5)
); 
Operadores — combineLatest
O operador combinado combineLatest é semelhante ao promise.all. Ele combina vários fluxos em um só. Assim que cada fluxo emite pelo menos uma vez, obtemos os últimos valores de cada um na forma de um array. Em seguida, após qualquer emissão dos fluxos combinados, ele retornará novos valores.
const { combineLatest, Observable } = Rx;
const { take } = RxOperators;
const observer_1 = Observable.create((observer) => {
let i = 1;
// Emite um valor a cada 1000ms
setInterval(() => {
observer.next('a: ' + i++);
}, 1000);
});
const observer_2 = Observable.create((observer) => {
let i = 1;
// Emite um valor a cada 750ms
setInterval(() => {
observer.next('b: ' + i++);
}, 750);
});
combineLatest(observer_1, observer_2).pipe(take(5));
Operadores — zip
Zip espera um valor de cada fluxo e forma um array baseado nesses valores. Se um valor não for recebido de um dos fluxos, o grupo não será formado.
const { zip, Observable } = Rx;
const { take } = RxOperators;
const observer_1 = Observable.create((observer) => {
let i = 1;
// Emite um valor a cada 1000ms
setInterval(() => {
observer.next('a: ' + i++);
}, 1000);
});
const observer_2 = Observable.create((observer) => {
let i = 1;
// Emite um valor a cada 750ms
setInterval(() => {
observer.next('b: ' + i++);
}, 750);
});
const observer_3 = Observable.create((observer) => {
let i = 1;
// Emite um valor a cada 500ms
setInterval(() => {
observer.next('c: ' + i++);
}, 500);
});
zip(observer_1, observer_2, observer_3).pipe(take(5));
Operadores — forkJoin
forkJoin também combina fluxos, mas ele emite um valor somente quando todos os fluxos estão completos.
const { forkJoin, Observable } = Rx;
const { take } = RxOperators;
const observer_1 = Observable.create((observer) => {
let i = 1;
// Emite um valor a cada 1000ms
setInterval(() => {
observer.next('a: ' + i++);
}, 1000);
}).pipe(take(3));
const observer_2 = Observable.create((observer) => {
let i = 1;
// Emite um valor a cada 750ms
setInterval(() => {
observer.next('b: ' + i++);
}, 750);
}).pipe(take(5));
const observer_3 = Observable.create((observer) => {
let i = 1;
// Emite um valor a cada 500ms
setInterval(() => {
observer.next('c: ' + i++);
}, 500);
}).pipe(take(4));
forkJoin(observer_1, observer_2, observer_3);
Operadores — map
O operador de transformação map converte o valor emitido em um novo.
const { Observable } = Rx;
const { take, map } = RxOperators;
Observable.create((observer) => {
let i = 1;
// Emite um valor a cada 1000ms
setInterval(() => {
observer.next(i++);
}, 1000);
}).pipe(
map(x => x * 10),
take(3)
);
Operadores — share, tap
O operador tap permite fazer efeitos colaterais, ou seja, realizar ações que não afetam a sequência.
O operador utilitário share é capaz de transformar um fluxo frio em um quente.
Terminamos com os operadores. Vamos passar para o Subject.
Pensando em voz alta
E aqui fui eu tomar um chá. Esses exemplos me cansaram 😀
Família de subjects
A família de subjects é um exemplo claro de fluxos quentes. Essas classes são uma espécie de híbrido que atuam simultaneamente como observable e observer. Como o subject é um fluxo quente, é necessário cancelar a inscrição. Falando sobre os principais métodos, temos:
- next – transmissão de novos dados no fluxo
- error – erro e finalização do fluxo
- complete – finalização do fluxo
- subscribe – inscrever-se no fluxo
- unsubscribe – cancelar a inscrição no fluxo
- asObservable – transforma em um observador
- toPromise – transforma em uma promessa
Diferenciam-se 4 a 5 tipos de subjects.
Pensando em voz alta
Na transmissão, mencionei 4, mas parece que adicionaram mais um. Como diz o ditado, viver é aprender.
Subject simples new Subject()– é o tipo mais simples de subject. É criado sem parâmetros. Transmite valores recebidos apenas após a inscrição.
BehaviorSubject new BehaviorSubject(defaultData) – na minha opinião, é o tipo mais comum de subjects. Recebe um valor padrão como entrada. Sempre mantém os dados da última emissão, que são transmitidos na inscrição. Esta classe também possui um método útil chamado value, que retorna o valor atual do fluxo.
ReplaySubject new ReplaySubject(bufferSize?: number, windowTime?: number) – Como entrada, pode opcionalmente aceitar o tamanho do buffer de valores que manterá como primeiro argumento e, como segundo, o tempo durante o qual as alterações são necessárias.
AsyncSubject new AsyncSubject() – ao se inscrever, nada acontece, e o valor será retornado apenas na finalização. Apenas o último valor do fluxo será retornado.
WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig | Observable, destination?: Observer) – A documentação não menciona nada sobre ele e eu o vejo pela primeira vez. Se alguém souber o que ele faz, escreva, vamos complementar.
Uf, bem, assim discutimos tudo o que eu queria falar hoje. Espero que esta informação tenha sido útil. Você pode consultar a lista de referências na aba de informações úteis.
Informação útil
- – link para a transmissão
- – Como funciona o JS: visão geral do motor, mecanismos de execução, pilha de chamadas
- — Como funciona o JS: ciclo de eventos, assíncrono e cinco maneiras de melhorar o código com async / await
- — Como funciona o Event Loop no JavaScript
- — A evolução do JavaScript assíncrono
- — O que é RxJS e por que é útil saber sobre ele
- — Aplicação prática do RxJS
- — Tutorial de Observables RxJS — Criando & Inscrevendo-se em Observables
- — RXJS: Observables Quentes e Frios
- — Classes, funções para criar Observable. Operadores.
- — Operadores RxJS por Exemplo
- — Lista de API
- — Tipos de Subject e Agendas no RxJS
Fonte: habr.com
