Olá a todos. Sergey Omelnitsky está em contato. Não faz muito tempo, hospedei um stream sobre programação reativa, onde falei sobre assincronia em JavaScript. Hoje gostaria de fazer anotações sobre este material.
Mas antes de iniciarmos o material principal, precisamos fazer uma nota introdutória. Então, vamos começar com as definições: o que é pilha e fila?
Pilha é uma coleção cujos elementos são obtidos com base no LIFO último a entrar, primeiro a sair
Fila é uma coleção cujos elementos são obtidos na base FIFO primeiro a entrar, primeiro a sair
Ok, vamos continuar.
JavaScript é uma linguagem de programação de thread único. Isso significa que há apenas um thread de execução e uma pilha na qual as funções são enfileiradas para execução. Portanto, o JavaScript só pode realizar uma operação por vez, enquanto outras operações aguardarão sua vez na pilha até serem chamadas.
Pilha de chamadas é uma estrutura de dados que, simplesmente, registra informações sobre o local do programa onde estamos. Se passarmos para uma função, colocamos sua entrada no topo da pilha. Quando retornamos de uma função, retiramos o elemento do topo da pilha e voltamos para onde chamamos a função. Isso é tudo o que a pilha pode fazer. E agora uma questão extremamente interessante. Como então a assincronia funciona em JavaScript?
Na verdade, além da pilha, os navegadores possuem uma fila especial para trabalhar com a chamada WebAPI. As funções nesta fila serão executadas em ordem somente depois que a pilha for completamente limpa. Somente depois disso eles são empurrados da fila para a pilha para execução. Se houver pelo menos um elemento na pilha no momento, eles não poderão ser adicionados à pilha. É justamente por isso que chamar funções por tempo limite muitas vezes não é preciso no tempo, uma vez que a função não pode passar da fila para a pilha enquanto estiver cheia.
Vejamos o exemplo a seguir e comecemos com sua implementação passo a passo. Vamos ver também o que acontece no sistema.
console.log('Hi');
setTimeout(function cb1() {
console.log('cb1');
}, 5000);
console.log('Bye');
1) Nada está acontecendo ainda. O console do navegador está limpo, a pilha de chamadas está vazia.
2) Em seguida, o comando console.log('Hi') é adicionado à pilha de chamadas.
3) E está cumprido
4) Em seguida, console.log('Hi') é removido da pilha de chamadas.
5) Agora passe para o comando setTimeout(function cb1() {… }). Ele é adicionado à pilha de chamadas.
6) O comando setTimeout(function cb1() {… }) é executado. O navegador cria um cronômetro que faz parte da API Web. Ele fará uma contagem regressiva.
7) O comando setTimeout(function cb1() {... }) concluiu seu trabalho e foi 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) Após pelo menos 5000 ms, o temporizador termina e coloca o retorno de chamada cb1 na fila de retorno de chamada.
12) O loop de eventos pega a função cb1 da fila de retorno de chamada 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.
Vejamos um exemplo em dinâmica:
Bem, vimos como a assincronia é implementada em JavaScript. Agora vamos falar brevemente sobre a evolução do código assíncrono.
A 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 a conhecemos em JavaScript só pode ser implementada por funções. Elas podem ser passadas como qualquer outra variável para outras funções. Foi assim que nasceram os retornos de chamada. E é legal, divertido e lúdico, até virar tristeza, melancolia e tristeza. Por que? É simples:
- À medida que a complexidade do código aumenta, o projeto rapidamente se transforma em blocos obscuros e repetidamente aninhados - “inferno de retorno de chamada”.
- O tratamento de erros pode ser fácil de ignorar.
- Você não pode retornar expressões com retorno.
Com o advento do Promise, 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);
});- Apareceram cadeias de promessas, o que melhorou a legibilidade do código
- Um método separado para detectar erros apareceu
- Adicionada a possibilidade de execução paralela usando Promise.all
- Podemos resolver a assincronia aninhada usando async/await
Mas as promessas têm suas limitações. Por exemplo, uma promessa não pode ser cancelada sem dançar com um pandeiro, e o mais importante é que funcione com um valor.
Bem, abordamos suavemente a programação reativa. Cansado? Bem, felizmente você pode ir tomar um chá, pensar sobre isso e voltar para ler mais. E eu continuarei.
Programação reativa é um paradigma de programação focado em fluxos de dados e propagação de mudanças. Vamos dar uma olhada mais de perto no 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. Estamos criando um array e para cada keyup do evento de entrada iremos armazenar o evento em nosso array. Ao mesmo tempo, gostaria de observar que nosso array é classificado por tempo, ou seja, o índice de eventos posteriores é maior que o índice dos anteriores. Essa matriz é um modelo simplificado de fluxo de dados, mas ainda não é um fluxo. Para que esse array seja chamado de stream com segurança, ele deve ser capaz de informar de alguma forma aos assinantes que novos dados chegaram nele. Assim chegamos à definição de fluxo.
Fluxo de dados
const { interval } = Rx;
const { take } = RxOperators;
interval(1000).pipe(
take(4)
)
Fluxo é uma matriz de dados classificados por hora que pode indicar que os dados foram alterados. Agora imagine como é conveniente escrever código em que uma ação requer a chamada de vários eventos em diferentes partes do código. Simplesmente assinamos o stream e ele nos notificará quando ocorrerem alterações. E a biblioteca RxJs pode fazer isso.
RxJS é uma biblioteca para trabalhar com programas assíncronos e baseados em eventos usando sequências observáveis. A biblioteca fornece um tipo básico Observável, vários tipos auxiliares (Observador, Agendadores, Assuntos) e operadores para trabalhar com eventos e coleções (mapear, filtrar, reduzir, cada e similares do JavaScript Array).
Vamos entender os conceitos básicos desta biblioteca.
Observável, Observador, Produtor
Observável é o primeiro tipo básico que veremos. Esta classe contém a parte principal da implementação do RxJs. Ele está associado a um fluxo observável, que pode ser assinado usando o método subscribe.
Observable implementa um mecanismo auxiliar para criar atualizações, o chamado Observador. A fonte de valores para o Observer é chamada Produtor. Pode ser um array, iterador, web socket, algum tipo de evento, etc. Portanto podemos dizer que observável é um condutor entre Produtor e Observador.
Observable lida com três tipos de eventos do Observer:
- próximo – novos dados
- erro – um erro se a sequência terminou devido a uma exceção. este evento também implica a conclusão da sequência.
- complete — sinaliza sobre a conclusão da sequência. Isso significa que não haverá mais dados novos.
Vamos ver a demonstração:
No início processaremos os valores 1, 2, 3 e após 1 segundo. obteremos 4 e encerraremos nosso stream.
Pensando alto
E então percebi que contar era mais interessante do que escrever sobre isso. 😀
Subscreva
Quando nos inscrevemos em um stream, criamos uma nova classe tudo inclusoo que nos dá a capacidade de cancelar a assinatura usando o método desinscrever. Também podemos agrupar assinaturas usando o método adicionar. Bem, é lógico que possamos desagrupar os threads usando remover. Os métodos add e remove aceitam outra assinatura como entrada. Gostaria de observar que, quando cancelamos a assinatura, cancelamos a assinatura de todas as assinaturas filhas como se elas tivessem chamado o método de cancelamento. Vá em frente.
Tipos de fluxos
HOT
FRIO
O produtor é criado fora do observável
O produtor é criado dentro do observável
Os dados são transferidos no momento em que o observável é criado
Os dados são fornecidos no momento da assinatura
Precisa de lógica adicional para cancelar a assinatura
O thread termina sozinho
Usa um relacionamento um-para-muitos
Usa um relacionamento um-para-um
Todas as assinaturas têm o mesmo significado
As assinaturas são independentes
Os dados podem ser perdidos se você não tiver uma assinatura
Reemite todos os valores de stream para uma nova assinatura
Para fazer uma analogia, eu pensaria em um fluxo quente como um filme no cinema. Em que momento você chegou, a partir desse momento você começou a assistir. Eu compararia um fluxo frio a uma chamada técnica. apoiar. Qualquer chamador ouve a gravação do correio de voz do início ao fim, mas você pode desligar cancelando a assinatura.
Gostaria de observar que também existem os chamados fluxos quentes (encontrei esta definição muito raramente e apenas em comunidades estrangeiras) - este é um fluxo que se transforma de um fluxo frio em um fluxo quente. Surge a questão - onde usar)) Vou dar um exemplo da prática.
Estou trabalhando com Angular. Ele usa ativamente rxjs. Para receber dados no servidor, espero um thread frio e uso esse thread no template usando asyncPipe. Se eu usar esse pipe várias vezes, então, voltando à definição de fluxo frio, cada pipe irá solicitar dados do servidor, o que é no mínimo estranho. E se eu converter um fluxo frio em quente, a solicitação acontecerá uma vez.
Em geral, entender o tipo de fluxo é bastante difícil 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 expandir nossa capacidade de trabalhar com fluxos. Eles ajudam a controlar os eventos que ocorrem no Observável. Veremos alguns dos mais populares e mais detalhes sobre as operadoras podem ser encontrados usando os links nas informações úteis.
Operadores - de
Vamos começar com o operador auxiliar de. Ele cria um Observable baseado em um valor simples.
Operadores - filtro
O operador de filtro, como o nome sugere, filtra o sinal do fluxo. Se o operador retornar verdadeiro, ele avança.
Operadores - pegue
take — Assume o valor do número de emissores, após o qual o thread termina.
Operadores - debounceTime
debounceTime - descarta os valores emitidos que estão dentro do intervalo de tempo especificado entre as saídas - após o intervalo de tempo ter passado, emite o último valor.
const { Observable } = Rx;
const { debounceTime, take } = RxOperators;
Observable.create((observer) => {
let i = 1;
observer.next(i++);
// Испускаем значение раз в 1000мс
setInterval(() => {
observer.next(i++)
}, 1000);
// Испускаем значение раз в 1500мс
setInterval(() => {
observer.next(i++)
}, 1500);
}).pipe(
debounceTime(700), // Ожидаем 700мс значения прежде чем обработать
take(3)
);
Operadores - takeWhile
Emite valores até que takeWhile retorne false, após o qual cancela a assinatura do thread.
const { Observable } = Rx;
const { debounceTime, takeWhile } = RxOperators;
Observable.create((observer) => {
let i = 1;
observer.next(i++);
// Испускаем значение раз в 1000мс
setInterval(() => {
observer.next(i++)
}, 1000);
}).pipe(
takeWhile( producer => producer < 5 )
);
Operadores - combineLatest
O operador combineLatest é um pouco semelhante a promessa.all. Ele combina vários threads em um. Depois que cada thread faz pelo menos uma emissão, obtemos os valores mais recentes de cada um na forma de um array. Além disso, após qualquer emissão dos fluxos mesclados, ele fornecerá novos valores.
const { combineLatest, Observable } = Rx;
const { take } = RxOperators;
const observer_1 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 1000мс
setInterval(() => {
observer.next('a: ' + i++);
}, 1000);
});
const observer_2 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 750мс
setInterval(() => {
observer.next('b: ' + i++);
}, 750);
});
combineLatest(observer_1, observer_2).pipe(take(5));
Operadores - zip
Zip – Espera por um valor de cada thread e forma um array baseado nesses valores. Se o valor não vier de nenhum thread, o grupo não será formado.
const { zip, Observable } = Rx;
const { take } = RxOperators;
const observer_1 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 1000мс
setInterval(() => {
observer.next('a: ' + i++);
}, 1000);
});
const observer_2 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 750
setInterval(() => {
observer.next('b: ' + i++);
}, 750);
});
const observer_3 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 500
setInterval(() => {
observer.next('c: ' + i++);
}, 500);
});
zip(observer_1, observer_2, observer_3).pipe(take(5));
Operadores - forkJoin
forkJoin também une threads, mas só emite um valor quando todos os threads são concluídos.
const { forkJoin, Observable } = Rx;
const { take } = RxOperators;
const observer_1 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 1000мс
setInterval(() => {
observer.next('a: ' + i++);
}, 1000);
}).pipe(take(3));
const observer_2 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 750
setInterval(() => {
observer.next('b: ' + i++);
}, 750);
}).pipe(take(5));
const observer_3 = Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 500
setInterval(() => {
observer.next('c: ' + i++);
}, 500);
}).pipe(take(4));
forkJoin(observer_1, observer_2, observer_3);
Operadores - mapa
O operador de transformação de mapa transforma o valor do emissor em um novo.
const { Observable } = Rx;
const { take, map } = RxOperators;
Observable.create((observer) => {
let i = 1;
// Испускаем значение раз в 1000мс
setInterval(() => {
observer.next(i++);
}, 1000);
}).pipe(
map(x => x * 10),
take(3)
);
Operadores – compartilhe, toque
O operador tap permite realizar efeitos colaterais, ou seja, quaisquer ações que não afetem a sequência.
O operador de utilidade pública compartilhada pode transformar um fluxo frio em quente.
Terminamos com os operadores. Vamos passar para o assunto.
Pensando alto
E então fui tomar um chá. Estou cansado desses exemplos 😀
Família sujeita
A família de sujeitos é um excelente exemplo de fluxos quentes. Essas classes são uma espécie de híbrido que atuam simultaneamente como observável e observador. Como o assunto é um tópico quente, é necessário cancelar sua inscrição. Se falarmos sobre os métodos principais, então estes são:
- próximo – transferência de novos dados para o fluxo
- erro – erro e encerramento do thread
- complete – conclusão do tópico
- inscrever-se – inscrever-se em um stream
- cancelar inscrição – cancelar inscrição no stream
- asObservable – transformar em um observador
- toPromise – transforma-se em uma promessa
Existem 4 5 tipos de assuntos.
Pensando alto
Havia 4 pessoas conversando na transmissão, mas acabaram acrescentando mais uma. Como se costuma dizer, viva e aprenda.
Assunto Simples new Subject()– o tipo mais simples de assuntos. Criado sem parâmetros. Transmite valores recebidos somente após assinatura.
ComportamentoAssunto new BehaviorSubject( defaultData<T> ) – na minha opinião, o tipo de assunto mais comum. A entrada assume o valor padrão. Salva sempre os dados da última edição, que é transmitida no momento da assinatura. Esta classe também possui um método de valor útil, que retorna o valor atual do fluxo.
Replay Subject new ReplaySubject(bufferSize?: number, windowTime?: number) — A entrada pode opcionalmente tomar como primeiro argumento o tamanho do buffer de valores que irá armazenar em si, e como segundo o tempo durante o qual precisamos de alterações.
AssinaturaAssunto new AsyncSubject() — nada acontece durante a assinatura e o valor será retornado somente quando concluído. Somente o último valor do stream será retornado.
WebSocketSubject new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig<T> | Observable<T>, destination?: Observer<T>) — A documentação é omissa sobre ele e estou vendo-o pela primeira vez. Se alguém souber o que ele faz, escreva e nós adicionaremos.
Ufa. Bem, cobrimos tudo o que eu queria contar a vocês hoje. Espero que esta informação tenha sido útil. Você mesmo pode ler a lista de referências na guia de informações úteis.
informações úteis
Fonte: habr.com