我认为我掌握了基本概念，但还存在一些模糊点。
总的来说，这是我如何使用 Observable：
observable.subscribe(x => {
})
如果我想筛选数据，可以使用这个：
import { first, last, map, reduce, find, skipWhile } from 'rxjs/operators';
observable.pipe(
map(x => {return x}),
first()
).subscribe(x => {
})
我也可以这样做：
import 'rxjs/add/operator/map';
import 'rxjs/add/operator/first';
observable.map(x => {return x}).first().subscribe(x => {
})
所以我的问题是：
pipe函数？"Pipeable"（之前称为“lettable”）操作符是自 RxJS 5.5 以来当前和推荐使用的方式。
我强烈建议您阅读官方文档中关于可管道化操作符的部分
主要区别在于它更容易制作自定义操作符，并且更好地支持树摇晃，而不会修改某些全局的Observable对象。如果两个不同的方面想要创建同名的操作符，则可能会发生冲突。
对每个操作符使用单独的import语句'rxjs/add/operator/first'是减小应用程序包的一种方法。通过仅导入所需的操作符而不是整个RxJS库，可以显着减少总包大小。然而编译器无法知道您是否导入'rxjs/add/operator/first'因为你真的需要它在你的代码中，还是只是在重构代码时忘记了将其删除。这是使用可管道化操作符的优势之一，在其中未使用的导入会自动被忽略。
rxjs-compat 包。https://github.com/ReactiveX/rxjs/blob/master/docs_app/content/guide/v6/migration.md#rxjs-v5x-to-v6-update-guide - martin管道方法
可管道的操作符是一个接收 observable 作为输入并返回另一个 observable 的函数。前一个 observable 保持不变。
pipe(...fns: UnaryFunction<any, any>[]): UnaryFunction<any, any>
管道是什么意思？
这意味着你之前在可观察对象实例上使用的任何操作符都可以作为纯函数在
rxjs/operators中使用。这使得构建操作符的组合或重用变得非常容易，而无需诸如创建扩展 Observable 的自定义可观察对象、然后覆盖 lift 等各种编程技巧来创建自定义的东西。
const { Observable } = require('rxjs/Rx')
const { filter, map, reduce, } = require('rxjs/operators')
const { pipe } = require('rxjs/Rx')
const filterOutWithEvens = filter(x => x % 2)
const doubleByValue = x => map(value => value * x);
const sumValue = reduce((acc, next) => acc + next, 0);
const source$ = Observable.range(0, 10)
source$.pipe(
filterOutWithEvens,
doubleByValue(2),
sumValue)
.subscribe(console.log); // 50
什么是不同之处？ 从您的示例中可以看出，主要区别在于提高源代码的可读性。您的示例中只有两个函数，但是想象一下如果有十几个函数呢？那么它将变成
function1().function2().function3().function4()
这真的很丑陋，阅读起来也很困难，特别是当您填写函数内部时。此外，某些编辑器（如Visual Studio code）不允许超过140行长度。但如果像以下方式那样处理，则：
Observable.pipe(
function1(),
function2(),
function3(),
function4()
)
这显著提高了可读性。
如果没有区别，为什么需要pipe()函数？PIPE()函数的目的是将所有接受和返回observable的函数合并在一起。它最初接受一个observable，然后在其内部使用每个函数使用的observable来执行pipe()函数。第一个函数接收observable，处理它、修改它的值，并将其传递给下一个函数，接着下一个函数接收第一个函数的输出observable，处理它，并传递到下一个函数，以此类推，直到pipe()函数中的所有函数都使用该observable，最终你会得到已处理的observable。最后，你可以使用subscribe()函数执行observable，从中提取出值。请记住，原始observable中的值不会被更改！！
为什么这些函数需要不同的导入？导入取决于函数在rxjs包中的指定位置。它是这样的：所有模块都存储在Angular的node_modules文件夹中。import { class } from "module";
让我们以以下代码为例。我刚刚在stackblitz中编写了它，所以没有自动生成任何内容，也没有从其他地方复制任何内容。当您可以直接阅读文档时，我认为将文档中陈述的内容复制过来没有意义。我假设您在这里提出此问题，是因为您没有理解文档。
Of()函数返回一个observable，在订阅时按顺序发出数字。
Observable还未被订阅。
当你使用Observable.pipe()时，pipe()函数将给定的Observable作为输入使用。
第一个函数map()使用该Observable进行处理，返回经过处理的Observable回到pipe()函数中，
然后，如果有下一个函数，它会将已处理的Observable传递给下一个函数，并且一直这样进行，直到所有函数都处理了Observable，
最后，pipe()函数将已处理的Observable返回到变量中，在以下示例中为obs。
现在Observable中的事情是，只要观察者没有订阅它，它就不会发出任何值。因此，我使用subscribe()函数订阅此Observable，一旦我订阅了它，of()函数就开始发出值，然后它们通过pipe()函数进行处理，最终你可以得到最终结果。例如，1从of()函数中获取，1在map()函数中加上1并返回。你可以将该值作为参数放入subscribe(function (argument) {})函数中获取。如果要打印它，请使用以下方式：
subscribe( function (argument) {
console.log(argument)
}
)
import { Component, OnInit } from '@angular/core';
import { pipe } from 'rxjs';
import { Observable, of } from 'rxjs';
import { map } from 'rxjs/operators';
@Component({
selector: 'my-app',
templateUrl: './app.component.html',
styleUrls: [ './app.component.css' ]
})
export class AppComponent implements OnInit {
obs = of(1,2,3).pipe(
map(x => x + 1),
);
constructor() { }
ngOnInit(){
this.obs.subscribe(value => console.log(value))
}
}
const compose = (fns) => fns.reduce((fn, composition) => (...args) => composition(fn(...args)))，2）使用！const composition = compose([fn4, ... , fn1])。如果你想要反向组合，还有额外的奖励：const pipe = compose(fns.reverse())。 - Nick Bullmap、first等构造可观察对象的方法。其他Rx的实现，例如RxJava，不需要管道函数。 - Max
pipe()函数能让你传递自己创建的操作符吗？ - zero298