🥇JavaScript 中的异步编程（回调、Promise、RxJs）| 专业主机

大家好。谢尔盖·奥梅尔尼茨基已与您联系。不久前，我主持了一个关于反应式编程的直播，其中我讨论了 JavaScript 中的异步。今天我想对这个材料做笔记。

但在开始主要材料之前，我们需要做一个介绍性说明。那么让我们从定义开始：什么是堆栈和队列？

叠放是一个集合，其元素是按照后进先出 LIFO 原则获取的

队列是一个集合，其元素是按照先进先出 FIFO 原则获取的

好吧，我们继续。

JavaScript 是一种单线程编程语言。这意味着只有一个执行线程和一个堆栈，函数在堆栈上排队等待执行。因此，JavaScript 一次只能执行一个操作，而其他操作将在堆栈上等待，直到被调用。

调用堆栈 是一种数据结构，简单来说，记录了我们在程序中所处位置的信息。如果我们传入一个函数，我们会将其条目推入堆栈顶部。当我们从函数返回时，我们从堆栈中弹出最顶层的元素，并最终回到我们调用该函数的位置。这就是堆栈所能做的全部事情。现在是一个非常有趣的问题。那么异步在 JavaScript 中是如何工作的呢？

事实上，除了堆栈之外，浏览器还有一个特殊的队列来处理所谓的 WebAPI。只有当堆栈被完全清除后，该队列中的函数才会按顺序执行。只有在此之后，它们才会从队列推入堆栈以供执行。如果此时堆栈上至少有一个元素，则无法将它们添加到堆栈中。正因为如此，通过超时调用函数往往在时间上不精确，因为函数在队列已满时无法从队列进入堆栈。

让我们看一下下面的示例并开始逐步实施。我们还可以看看系统中发生了什么。

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

1）什么都没有发生。浏览器控制台是清晰的，调用堆栈是空的。

2) 然后将命令 console.log('Hi') 添加到调用堆栈中。

3）并且它已经实现了

4) 然后 console.log('Hi') 从调用堆栈中删除。

5) 现在继续执行命令 setTimeout(function cb1() {… })。它被添加到调用堆栈中。

6) 执行setTimeout(function cb1() {… })命令。浏览器创建一个计时器，它是 Web API 的一部分。它将执行倒计时。

7) setTimeout(function cb1() {... }) 命令已完成其工作并从调用堆栈中删除。

8) 命令 console.log('Bye') 被添加到调用堆栈中。

9) 执行console.log('Bye')命令。

10) 命令 console.log('Bye') 已从调用堆栈中删除。

11) 至少经过 5000 ms 后，计时器终止并将回调 cb1 放入回调队列中。

12) 事件循环从回调队列中获取函数 cb1 并将其放置在调用堆栈上。

13) 执行函数 cb1 并将 console.log('cb1') 添加到调用堆栈中。

14) 执行命令console.log('cb1')。

15) 命令 console.log('cb1') 已从调用堆栈中删除。

16) 函数cb1 从调用堆栈中删除。

我们来看一个动力学中的例子：

好吧，我们研究了如何在 JavaScript 中实现异步。现在我们简单谈谈异步代码的演变。

异步代码的演变。

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

正如我们所知，JavaScript 中的异步编程只能通过函数来实现。它们可以像任何其他变量一样传递给其他函数。这就是回调的诞生方式。而且很酷、很好玩、很好玩，直到变成悲伤、惆怅、悲伤。为什么？这很简单：

随着代码复杂性的增加，项目很快就会变成晦涩的、重复嵌套的块——“回调地狱”。
错误处理很容易被忽略。
不能用 return 返回表达式。

随着Promise的出现，情况变得好一些了。

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

Promise 链出现，提高了代码可读性
出现了一种单独的捕获错误的方法
添加了使用 Promise.all 并行执行的可能性
我们可以使用 async/await 解决嵌套异步问题

但承诺也有其局限性。例如，如果不敲着手鼓跳舞，就无法取消承诺，最重要的是它符合一个值。

好吧，我们已经顺利地接近了响应式编程。疲劳的？好吧，幸运的是，你可以去泡杯茶，想一想，然后再回来阅读更多内容。我会继续。

反应式编程 是一种专注于数据流和变更传播的编程范例。让我们仔细看看什么是数据流。

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

const eventsArray = [];

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

假设我们有一个输入字段。我们正在创建一个数组，对于输入事件的每个按键，我们都会将该事件存储在数组中。同时，我想指出的是，我们的数组是按时间排序的，即较晚事件的索引大于较早事件的索引。这样的数组是数据流的简化模型，但它还不是流。为了使该数组能够安全地称为流，它必须能够以某种方式通知订阅者新数据已到达其中。这样我们就得出了流的定义。

数据流

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

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

流是按时间排序的数据数组，可以指示数据已更改。现在想象一下，编写一个操作需要在代码的不同部分调用多个事件的代码会变得多么方便。我们只需订阅该流，当发生变化时它会通知我们。 RxJs 库可以做到这一点。

接收JS 是一个使用可观察序列处理异步和基于事件的程序的库。该库提供了基本类型 可观察的，几种辅助类型（观察者、调度者、受试者) 和用于像处理集合一样处理事件的运算符 (映射、过滤、减少、每个 以及 JavaScript 数组中的类似内容）。

让我们了解一下这个库的基本概念。

可观察者、观察者、生产者

Observable 是我们要讨论的第一个基本类型。此类包含 RxJs 实现的主要部分。它与一个可观察流关联，可以使用 subscribe 方法订阅该流。

Observable 实现了一种用于创建更新的辅助机制，即所谓的 观察员。观察者的值的来源称为 制片人。这可以是数组、迭代器、Web 套接字、某种事件等。所以我们可以说observable是Producer和Observer之间的导体。

Observable 处理三种类型的观察者事件：

下一步 – 新数据
error – 如果序列由于异常而结束，则会出现错误。此事件也意味着序列的完成。
完成——关于序列完成的信号。这意味着不会有更多新数据。

让我们看一下演示：

一开始我们将处理值1、2、3以及1秒后。我们将得到 4 并结束我们的流。

把想法大声说出来

然后我意识到讲述它比写它更有趣。 😀

订购

当我们订阅流时，我们创建一个新类订阅这使我们能够使用该方法取消订阅退订。我们还可以使用该方法对订阅进行分组加。好吧，我们可以使用以下命令取消线程分组，这是合乎逻辑的去掉。添加和删除方法接受另一个订阅作为输入。我想指出的是，当我们取消订阅时，我们会取消所有子订阅的订阅，就像它们调用了取消订阅方法一样。前进。

流的类型

热卖
冷

Producer 是在 observable 外部创建的
生产者是在 observable 内部创建的

数据在创建可观察对象时传输
数据在订阅时提供

需要额外的逻辑来取消订阅
线程自行终止

使用一对多关系
使用一对一关系

所有订阅都具有相同的含义
订阅是独立的

如果您没有订阅，数据可能会丢失
重新发出新订阅的所有流值

打个比方，我会把热流想象成电影院里的电影。你在什么时间到达，从那一刻起你就开始观看。我将冷流与技术中的呼叫进行比较。支持。任何呼叫者都会从头到尾收听语音邮件录音，但您可以使用取消订阅挂断电话。

我想指出的是，还有所谓的暖流（我很少遇到这个定义，仅在国外社区） - 这是一种从冷流转变为热流的流。问题出现了-在哪里使用））我将给出一个实践中的例子。

我正在使用 Angular。他积极使用 rxjs。为了将数据接收到服务器，我期望一个冷线程并使用 asyncPipe 在模板中使用该线程。如果我多次使用这个管道，那么，回到冷流的定义，每个管道都会向服务器请求数据，这至少可以说很奇怪。如果我将冷流转换为热流，则该请求将发生一次。

一般来说，理解流的类型对于初学者来说相当困难，但很重要。

运营商

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

操作符使我们能够扩展我们使用流的能力。它们帮助控制 Observable 中发生的事件。我们将研究几个最流行的操作符，并且可以使用有用信息中的链接找到有关操作符的更多详细信息。

运算符 - of

让我们从辅助运算符 of 开始。它基于一个简单的值创建一个 Observable。

运算符 - 过滤器

过滤器运算符，顾名思义，过滤流信号。如果运算符返回 true，则进一步跳过。

运营商-采取

take — 获取发射器数量的值，之后线程结束。

运算符 - debounceTime

debounceTime - 丢弃在输出之间的指定时间间隔内发出的值 - 在时间间隔过去后，发出最后一个值。

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

运算符 - takeWhile

发出值直到 takeWhile 返回 false，之后取消订阅线程。

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

运营商-combineLatest

mergeLatest 操作符有点类似于promise.all。它将多个线程合二为一。每个线程至少进行一次发射后，我们以数组的形式从每个线程获取最新值。此外，在合并流发出任何数据后，它将给出新的值。

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

运营商 - zip

Zip - 等待来自每个线程的值并根据这些值形成一个数组。如果该值不是来自任何线程，则不会形成该组。

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

运算符 - forkJoin

forkJoin 也连接线程，但它仅在所有线程完成时发出一个值。

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

运营商 - 地图

地图变换运算符将发射器值变换为新的值。

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

运营商 – 分享、点击

Tap 运算符允许您执行副作用，即任何不影响序列的操作。

共享公用事业运营商可以将冷流变成热流。

我们与运营商的关系已经结束了。让我们继续主题。

把想法大声说出来

然后我就去喝茶了。我厌倦了这些例子😀

对象家庭

该主题族是热流的一个典型例子。这些类是一种混合体，同时充当可观察者和观察者。由于主题是热门话题，因此需要取消订阅。如果我们谈论主要方法，那么这些是：

next – 将新数据传输到流
error – 错误和线程终止
完成——线程的完成
订阅 – 订阅流
取消订阅 – 取消订阅流
asObservable – 转变为观察者
toPromise——转化为承诺

有4 5种科目。

把想法大声说出来

直播里本来有4个人在说话，结果又加了一个。正如他们所说，生活和学习。

简单主题 new Subject()– 最简单的科目类型。创建时不带参数。仅传输订阅后收到的值。

行为主体 new BehaviorSubject( defaultData<T> ) – 在我看来，这是最常见的主题类型。输入采用默认值。始终保存订阅时传输的最新一期数据。该类还有一个有用的 value 方法，它返回流的当前值。

重播主题 new ReplaySubject(bufferSize?: number, windowTime?: number) — 输入可以选择将其本身存储的值缓冲区的大小作为第一个参数，将我们需要更改的时间作为第二个参数。

异步主题 new AsyncSubject() — 订阅时不会发生任何事情，只有完成后才会返回值。仅返回流的最后一个值。

WebSocket主题 new WebSocketSubject(urlConfigOrSource: string | WebSocketSubjectConfig<T> | Observable<T>, destination?: Observer<T>) — 文档中没有提及他，而我还是第一次见到他。如果有人知道他做了什么，请写下来，我们会添加它。

唷。好吧，我们今天已经涵盖了我想告诉你的所有内容。我希望这些信息有用。您可以在有用信息选项卡中自行阅读参考文献列表。

有用的信息

来源： habr.com

JavaScript 中的异步编程（回调、Promise、RxJs）

异步代码的演变。

数据流

可观察者、观察者、生产者

订购

流的类型

运营商

对象家庭

有用的信息

添加评论 取消回复

添加评论取消回复