自从你创建这个问题以来，我一直在我的选项卡中打开这个问题，想要提供帮助，但也想要学习足够多的 RxJS，以便我自己创建一个解决方案。
非常抱歉，我没有查看您的示例，而是创建了自己的示例。我不得不请你原谅我在这里提供的非常大的答案。
我的主要动力来自 Ben Lesh 的精彩演讲，他是现代 RxJS 的创造者之一，你可以找到 here。我强烈建议您观看此视频，甚至多次观看，以尝试了解我在解决此问题时使用的一些内容。
和 Ben 一样，我也使用 Angular 框架作为这个项目的基础。您可以在GitHub 找到我的解决方案。此外，就像 Ben 在他的演讲中多次解释一些 Angular 特定的东西一样，我也会在这里尝试这样做。
我的应用中有一个简单的FeedComponent 和一个FeedService。该服务正在使用 Angular Dependency Injection 注入到FeedComponent。
现在，我有一个绑定到 FeedComponent that looks like this 的 HTML：
<div class="loading" *ngIf="loading$ | async">
Loading...
</div>
<div class="filter">
<form #form="ngForm">
...
</form>
</div>
<div #articles class="articles">
<article *ngFor="let article of feed$ | async">
...
</article>
</div>
您可以看到我有三个部分：<div> 负责显示正在加载提要的消息；另一个<div>，带有过滤<form>，显示过滤选项；第三个<div> 负责将提要项显示为<article>s 的列表。
这里的 Angular 特定内容包括 *ngIf 和 *ngFor 指令和 async 管道 (|)。一句话：*ngIf如果属性值中的条件为真，则渲染某个DOM元素； *ngFor 循环遍历提供的项目数组并渲染特定 DOM 元素的次数为数组长度；管道 | 转换项目，以便左侧的项目始终是右侧管道的输入项目，async 管道也是如此 - 它转换一个 Observable（您可以通过 $ 符号判断它是一个 Observable我和许多其他人正在使用的后缀）将来自 Observable 的项目转换为指令理解的项目。 async 管道在 Ben 的 talk 中也有解释。
让我们开始吧：你可以看到我在我的 HTML 模板中使用了两个 Observable，这就是你所需要的。一个会为您提供FeedItems 的数组，以便您可以在页面上显示它们，另一个会在从服务器获取提要时发出布尔值。如果您不使用 Angular，而是使用其他框架或库，或者根本不使用，您仍然可以只有这两个流。您必须手动订阅它们（并在以后不再需要时取消订阅），并且当您获得结果时，您应该相应地更新 DOM。 Angular 和 async pipe 在这里为我完成了所有这些工作。
这两个分别是feed$ 和loading$ Observable 流。它们都在绑定到这个 HTML 的FeedComponent 中定义，非常简单，像这样：
feed$ = this.feedService.feed$;
loading$ = this.feedService.loading$.pipe(delay(10));
正如我所说，feedService 使用 Angular DI 通过FeedComponent 的构造函数注入到FeedComponent：
constructor(private feedService: FeedService) {}
如果您要使用自己的 JS 框架/lib 或根本不使用任何 lib，则只需创建新的 FeedService 对象。我正在向feedService.loading$ 流添加 10 毫秒的延迟，因为我遇到了一些我不应该在这里解释的 Angular 错误。如果不使用 Angular，你可能根本不需要它。
现在，为了能够通过 feed$ 流提供提要项 (FeedItems[])，您需要监听两个可能的事件：当用户滚动到底部时触发的滚动事件页面和过滤器表单输入值更改时发生的事件。这两个事件需要组合成一个单独的 Observable，我们将其称为 filterSeed$ - 它会发出包含在使用的 form 的输入元素中的值。
第一个事件流可以由这两个 Observable 组成：
scrollPercent$: Observable<number> = fromEvent(document, 'scroll')
.pipe(
map(() => {
const scrollTop = this.articles.nativeElement.getBoundingClientRect().top;
const docHeight = this.articles.nativeElement.getBoundingClientRect().height;
const winHeight = window.innerHeight;
const scroll = scrollTop / (winHeight - docHeight);
return Math.round(scroll * 100);
})
);
loadMore$: Observable<number> = this.scrollPercent$
.pipe(
filter(percent => percent >= 80),
take(1),
repeatWhen(() => this.feedLoadingStops$)
);
scrollPercent$ 是一个发出一些数字的 Observable。它们代表document 的scroll 事件触发时的滚动百分比（使用fromEvent 创建）。无论它发出什么事件，我都不在乎它。我只关心它何时发出，所以我可以使用一些简单的数学将其map 转换为百分比。 this.articles.nativeElement 是 Angular 特定的，所以如果您需要另一个示例，请查看 this Pen 了解如何使用 jQuery 实现它。 map函数的返回值是四舍五入的滚动百分比。
loadMore$ 是一个 Observable，它仅在用户滚动到足以加载新的提要项时触发事件 - 它触发滚动百分比数字，但我们并不真正关心这一点，你会看到我们'稍后忽略这些数字。应该发生这种情况的阈值是滚动的 80% 或之后。所以，我在这里使用filter 只让那些高于阈值的项目（请记住，当达到并通过此阈值时，我需要loadMore$ 发出）。我在这里使用take(1)，因为我真的只需要一个这样的项目。
就像 Ben 拥有 had a problem 时整个流只工作一次一样，我也拥有它。因为takecompletes（有效地取消订阅）从源获取该项目时，我需要再次重新订阅相同的源，即，一直到顶部，由fromEvent 创建的流。
所以，我需要再次开始收听scroll 事件，但这里有一个问题：我不想立即开始，而是在加载完成时开始。所以，我需要使用repeatWhen 而不仅仅是repeat。 repeatWhen 采用工厂函数，它在需要订阅 Observable 时调用该函数。它监听提供的 Observable (this.feedLoadingStops$) 并在 this.feedLoadingStops$ 发出时重新订阅源。
this.feedLoadingStops$ 看起来像这样：
feedLoadingStops$ = this.loading$.pipe(map(v => !v), filter(v => v));
它将false 的值反转为true，反之亦然，因此发出的true 值表示加载已停止（请记住，当loading$ 发出false 时，表示加载已停止）。它还只过滤 true 值，以便我们仅在它停止加载时才获得发射。
但是，您可能会想知道为什么。为什么repeat 为 Ben 工作而不为我工作？为什么我必须在加载停止时重新订阅。这是因为我们在fromEvent 触发的事件之后都使用了higher order mapping operators，以便稍后将HTTP 请求展平。稍后我肯定会谈到这一点，但他使用的是exhaustMap 运算符，而我使用了switchMap，它总是切换到源最新发出的项目并订阅它。当loadMore$ 再次被重新订阅时（仅使用repeat），它将再次开始监听滚动事件，并且由于用户肯定会继续滚动更多，loadMore$ 将再次开始发射，switchMap 将继续一直重新订阅提供的 HTTP 请求，直到用户不会停止滚动。这真的不是我们想要的——我们不想仅仅因为用户正在做我们有责任解决的事情，就创建多个完全相同的 HTTP 请求来获取单个资源。 exhaustMap 是不同的，因此它与 swithcMap 完全相反 - 它会等待第一个发出的项目完成（基本上，它会耗尽），直到它订阅下一个。 p>
这就是第一个有助于创建filterSeed$ Observable 的事件流的解释。另一种比较简单。当涉及到表单时，Angular 自己提供了这样的 Observables。我使用了this.form.valueChanges，并且自动订阅了任何表单输入元素值更改。由于我只有一个<select>，我用它来伪造我应该加载包含所有项目或仅包含文本的项目或仅包含图像的项目的提要，我想听听用户何时选择不同的选项来触发事件.
由于 Angular 为我提供了此功能，您可能希望创建自己的 Observable，它会为您发出更改的表单输入值。
最后，这就是filterSeed$ 的样子：
filterSeed$: Observable<FeedFilter> = defer(() => merge(
this.loadMore$.pipe(map(() => this.form.value)),
this.form.valueChanges
));
在这里，我想merge 两个流：一个是通过侦听loadMore$ 事件创建的，另一个是侦听过滤器表单值更改的流。这正是我想要的：我只想在用户滚动到底部或更改过滤器时才加载新的提要项。 this.form.valueChanges 已经提供 FeedFilter 项目，但 this.loadMore$ 没有。请记住，this.loadMore$ 发出的数字让我很难过我并不真正关心，所以我将它们映射到 this.form.value。这又是 Angular 特定的，因此您必须实现自己对整个表单输入元素的读取。当前this.form.value 始终与this.form.valueChanges 中最后发出的项目相同。
这里使用defer 的原因是另一个特定的Angular，因为在创建filterSeed$ 时，this.form 仍然是undefined，所以我需要等到订阅发生以确保this.form 是可用的。我将在 ngAfterViewInit() 生命周期钩子中订阅这个 Observable，它在组件的视图创建时被调用。如果不使用 Angular，您可能不需要在此处使用 defer。现在，当一切准备就绪后，就可以订阅this.filterSeed$ 并进行一些数据加载了。不要忘记在离开组件/页面时取消订阅，以免通过不删除由fromEvent 创建的所有事件侦听器来泄漏内存。
ngAfterViewInit(): void {
this.subscription = this.filterSeed$.subscribe(this.feedService.filter$);
}
ngOnDestroy(): void {
this.subscription.unsubscribe();
}
我们现在来到这个答案的第二部分，即FeedService。 FeedService 只是一个简单的 JS 对象，它有一些 Observables 和一些状态。我需要这种状态才能使用多个 Observables——有些人可能会说这不是真正的 Rx 方式，但我发现这样解决它对我来说更容易。我还将Angular的HttpClient（作为http变量）注入到FeedService，它只是XHR的一个Angular包装器。您可以改用 RxJS ajax 静态创建方法 - 两者的行为应该相同。
FeedService 有三个公共 Observable，您可能已经看到它们都被用于 FeedComponent：filter$、loading$ 和 feed$ Observable。后面两个被FeedComponent 用来向DOM 渲染一些东西，而filter$ 被用来给它提供filterSeed$。基本上，filter$ 只是一个主题：
filter$ = new Subject<FeedFilter>();
而且由于 Subjects 既是 Observables 又是 Observers，我可以将它用作 Observer，所以我在订阅 FeedComponent 中的 filterSeed$ 时将它传递给了 subscribe 方法。这意味着filter$ Observer 将订阅filterSeed$，并且从filterSeed$ 对next 方法的任何调用（基本上是任何发射）都将传递给filter$ 主题。这意味着使用filter$ 的任何其他人都将获得filterSeed$ 发出的值。
而filter$ 用于创建feed$ Observable。这是它的样子：
feed$: Observable<FeedItem[]> = this.filter$.pipe(
switchMap(filter => {
if (filter !== this.filter) {
this.filter = filter;
this.nextPage = 1;
this.shouldReset = true;
}
return this.getFeed$;
}),
scan((acc, value) => {
if (this.shouldReset) {
this.shouldReset = false;
return value;
}
return acc.concat(value);
}, [])
)
我在这里使用了两个运算符：switchMap 和 scan。我也有一些状态，我在变量this.filter、this.nextPage 和this.shouldReset 中保留在类本身中。
我已经提到了higher order mapping operators。我在这里使用switchMap。它在filter$ Subject 触发的事件之后使用，该主题通过subscribe 方法与FeedComponent 的filterSeed$ 连接。因此，每当触发刷新事件（通过用户滚动足够多或通过用户更改过滤器表单中的过滤器）时，我想将其映射到负责创建 HTTP 请求的getFeed$ Observable。选择 switchMap 而不是其他人（而不是 Ben 使用的 exhaustMap）的原因是我想确保我总是从用户使用的最新过滤器中获得结果。 IE。如果用户在第一个请求仍在加载时发送了一个带有一个过滤器的请求并同时更改了一个过滤器，我想取消该请求并切换到另一个 HTTP 请求。
由于filter$ Subject 正在发出FeedFilter 对象，它们将作为filter 参数传递给switchMap 的回调函数。这是我检查filter 是否实际上与FeedService (this.filter) 中的过滤器相同的地方。如果不一样，说明过滤器变了，所以我需要将新过滤器保存到FeedService的过滤器（this.filter = filter;）。这也意味着我必须将页面重置为第 1 页（this.nextPage = 1;）并将this.shouldReset 设置为true。然后我返回getFeed$ 内部订阅的switchMap。
所有这些状态保持变量稍后被getFeed$ Observable 或switchMap 之后的下一个运算符使用：scan。但是，getFeed$ 看起来如何。方法如下：
getFeed$: Observable<FeedItem[]> = defer(() => {
if (this.nextPage) {
this.loadingSubject$.next(true);
const url = appendQuery('/feed', { page: this.nextPage, feedFilter: this.filter.feedFilter });
return this.http.get<FakeFeedResponse>(url);
} else {
return NEVER;
}
}).pipe(
catchError(() => /* Potentially handle this.nextPage here */EMPTY),
tap(response => {
this.nextPage = response.nextPage;
this.loadingSubject$.next(false);
}),
map(response => response.items),
share()
);
我在这里再次使用defer。这是因为我在这些流之外保存了一些状态，所以我希望在订阅 getFeed$ 时读取这些状态变量的值，而不是在创建 getFeed$ 对象时读取。
在defer 的回调函数体中，我正在检查this.nextPage 是否存在。如果没有更多要加载的项目，服务器将返回null，所以在这种情况下，我将返回NEVER，这是一个 从不 发出的 Observable。但是，如果有 项要加载（当 nextPage 是一个有效的真实数字时），我将返回一个 HTTP get 请求。 nextPage 默认设置为 1（或者如果过滤器更改，则在 switchMap 中重置为 1）。我正在通过appending nextPage 和feedFilter 构造url 作为'/feed' 路由的查询字符串。
另外，我使用loadingSubject$ 发出true，表示加载已经开始。 loadingSubject$ 看起来像这样：
private loadingSubject$ = new BehaviorSubject(false);
loading$: Observable<boolean> = this.loadingSubject$.asObservable();
这是一个BehaviorSubject，默认值为false。此主题发出的值提供给 FeedComponent 到 loading$ Observable。当FeedComponent第一次订阅loading$ Observable时，会立即得到false。
返回到defer 的this.http.get<FakeFeedResponse>(url) 请求正在使用Angular XHR 包装器。我将http 注入到FeedService，但您应该能够使用RxJS 的ajax，正如我已经提到的。它发出FakeFeedResponse 类型的对象。
在使用defer 构造了这个 Observable 之后，我想在它发出时做更多的事情。首先是使用catchError 处理错误。如果发生错误，我想返回一个 EMPTY Observable，它只是完成而不发出任何项目。我在这里添加了评论，以便您可以添加更多错误处理或处理（重新）设置nextPage 或其他内容。
在那之后，我将response 中的nextPage 保存到tap 中，同时将false 发送到loadingSubject$，表示加载已停止。在tap 之后，我使用map 从response 中提取items。
然后我 shareing 它。在我的情况下，这实际上并不是真正需要的。为什么？因为getFeed$（即switchMap）只有一个订阅者，所以实际上没有必要在多个订阅者之间使用share，但它可以留在这里如果它永远需要 - 实际上，如果存在另一个订阅者。我添加它是因为 Ben 也添加了它，但他的getFeed$ Observable 的订阅者不止一个。
这就是getFeed$ 在其回调方法中返回给switchMap 的全部内容。在feed$ Observable 中，在switchMap 之后，我使用scan。实际上，scan 在这里只是为了让 Angular 的 async 管道可以通过将新值连接到已加载的值 (acc.concat(value)) 来接收已加载的项目。我在这里使用this.shouldReset 标志，以便在更改过滤器时不使用concat。如果您不使用 Angular，您可能会自己订阅 feed$ Observable，并且您可能会自己处理这种情况，而不是 scan，因此您可能不需要在这里拥有 scan。
毕竟，我使用 Angular 拦截器功能来fake 所有服务器响应。请看看如何。
就是这样。很抱歉回答的很长，如果您有任何问题，请在 GitHub 上打开一个问题。我真的希望这个答案可以帮助您制定解决方案，我没有真正看过，对不起。
如果你想试试这个例子，你可以克隆项目，运行npm install，然后运行ng serve，这将编译整个项目并运行一个开发服务器，这样你就可以自己尝试这个项目了。