Angular / Rxjs - 排队 / 阻塞 HTTP 请求，直到前一个请求完成答案

【问题标题】：Angular / Rxjs - queuing / blocking HTTP requests until a previous request completesAngular / Rxjs - 排队 / 阻塞 HTTP 请求，直到前一个请求完成
【发布时间】：2018-03-29 15:02:27
【问题描述】：

我正在与使用一次性身份验证令牌的 API 服务器交谈，每次请求都会重新发布（即生成新令牌）。如果使用同一个令牌同时发出两个 API 请求，一个会由于令牌过期而失败，并且可能会不必要地触发错误处理程序。

我想做的是建立一种等待机制，这样一次只发出一个 HTTP 请求，后续请求排队等待前一个请求完成。

我想以一种通用的方式构建它，这样我就可以有一个服务层来发出请求，因此它对上层是透明的，但是我正在努力寻找一种巧妙的方法来做到这一点。有什么建议吗？

-- edit --

我应该提到，所有后续 API 调用 URL/请求正文都直接依赖于先前 API 调用的响应，例如GET /books?apiKey=ABC 返回：

{ 'apiKey': 'XYZ', 'names': [...] }

排队的下一个请求必须等待此响应并附加新的 apiKey：GET /authors?apiKey=XYZ

在上面的例子中，执行GET /authors?apiKey=ABC 会导致错误。

如果两个 HTTP 请求之间存在很强的依赖关系（例如请求书，然后根据其 id 请求该特定书的作者），则可以使用 flatMap、嵌套订阅等对它们进行序列化。我需要与此类似的功能，但以一种通用的方式，我可以将请求添加到在运行时序列化的队列中。

-- edit two --

我有两个组件，如下所示，假设它们都在启动时调用 API。

class ComponentA { 
    ngOnInit(api:<BookApi extends CommonApi>) {
        this.api.list().subscribe(x => {...});
    }
}

class ComponentB { 
    ngOnInit(api:<CityLookupApi extends CommonApi>) {
        this.api.list().subscribe(x => {...});
    }
}

所有 API 服务都扩展了一个通用 API 以隐藏其背后的 apiKey 处理，因此组件可以使用更高级别的调用来帮助理解。问题在于，如果两个组件几乎同时初始化，则两个调用都可以使用相同的 api 密钥，其中一个会失败。但是，我无法使用例如创建这些 Observable 批次。 forkJoin 因为它们是在运行时异步创建和订阅的，因为它们是在各自的组件中创建的。

【问题讨论】：

标签： angular rxjs

【解决方案1】：

Yanis-git 解决了排队问题。如果您想在其他请求发生时阻止所有请求，请使用exhaustMap。

exhaustmap docs

【讨论】：

【解决方案2】：

您可以使用Observable.concat(...myArrayOfAjaxRequestObservable) 让每个 observable 一个一个地消耗。

只有 Observable.contact 必须自己订阅您的 ajax 请求。只有这样才能 100% 确定请求不会被代码的另一部分触发。

---更新---

示例：https://stackblitz.com/edit/angular-66ic6c?file=app%2Fapp.component.ts

【讨论】：

这可能是 OP 的方式。我添加了排气映射答案作为某些边缘情况的替代解决方案。投赞成票:)
嗨 @TomaszKula 我不确定为什么我们需要 exhaustMap 和 Observable.concat。正如您在我附加的示例中看到的那样，所有请求都在逐步消耗，并且在未收到当前答案之前一切都被阻止。
.concat 不需要它。它是一个或另一个。这是一种替代解决方案。
好的，很好的说明和愉快的编码。
这似乎不太像我需要的。这会将请求排队并一一等待响应，但它仍然会同时发出所有请求。我将您的示例更改为根据 API 响应中的“种子”附加一个 apiKey 参数。如果您检查网络选项卡，它将显示所有请求都使用相同的密钥，而不是在触发下一个请求之前等待插入新种子/密钥的响应：stackblitz.com/edit/angular-sx4zxx?file=app%2Fapp.component.ts

【解决方案3】：

我会考虑将mergeMap 与其concurrency 参数一起使用。

这是一个模拟我想要的代码的 sn-p

import {Observable} from 'rxjs';
import {Subject} from 'rxjs';

const requestsRemoteService = new Subject<string>();
const requestStream = requestsRemoteService.asObservable();

// this function simulates the execution of the request
const requestExecution = (input: string) => {
    return Observable.of(input + ' executed').delay(2000);
}

// this is what the service should to queue the requests
requestStream
// the last parameter, set to 1, is the level of concurrency
.mergeMap(requestInput => requestExecution(requestInput), 1)
.subscribe(
    data => console.log(data),
    err => console.error(err),
    () => console.log('DONE')
)

// these are the various requests coming to the service
setTimeout(() => requestsRemoteService.next('First request'), 1);
setTimeout(() => requestsRemoteService.next('Second request'), 2);
setTimeout(() => requestsRemoteService.next('Third request'), 3);

【讨论】：