ReactiveX 的各种语言特定实现提供了一系列操作符，您可以使用它们将 Observable 或 Observable 发射的一系列项目转换为另一个对象或数据结构。其中一些操作符会阻塞，直到 Observable 终止，然后生成等效的对象或数据结构；另一些操作符会返回一个 Observable，该 Observable 会发射这样的对象或数据结构。
在 ReactiveX 的一些实现中，还存在一个操作符，它可以将 Observable 转换为一个“Blocking” Observable。Blocking Observable 扩展了普通 Observable，它提供了一组方法，这些方法在 Observable 发射的项目上进行操作，并会阻塞。一些 To 操作符位于此 Blocking Observable 扩展操作集。
getIterator 操作符适用于 BlockingObservable 子类，因此为了使用它，您必须首先使用 BlockingObservable.from 方法或 Observable.toBlocking 操作符将源 Observable 转换为 BlockingObservable。
此操作符将 Observable 转换为一个 Iterator，您可以使用它迭代源 Observable 发射的项目集。
ReactiveXBlockingObservable.getIterator()
toFuture 操作符适用于 BlockingObservable 子类，因此为了使用它，您必须首先使用 BlockingObservable.from 方法或 Observable.toBlocking 操作符将源 Observable 转换为 BlockingObservable。
此操作符将 Observable 转换为一个 Future，它将返回源 Observable 发射的单个项目。如果源 Observable 发射了多个项目，Future 将收到一个 IllegalArgumentException；如果它在发射任何项目后完成，Future 将收到一个 NoSuchElementException。
如果您想将一个可能发射多个项目的 Observable 转换为 Future，请尝试以下操作：myObservable.toList().toBlocking().toFuture()。
BlockingObservable.toFuture()
toIterable 操作符适用于 BlockingObservable 子类，因此为了使用它，您必须首先使用 BlockingObservable.from 方法或 Observable.toBlocking 操作符将源 Observable 转换为 BlockingObservable。
此操作符将 Observable 转换为一个 Iterable，您可以使用它迭代源 Observable 发射的项目集。
BlockingObservable.toIterable()
通常，发射多个项目的 Observable 会为每个这样的项目调用其观察者的 onNext 方法。您可以更改此行为，指示 Observable 组成一个包含这些多个项目的列表，然后仅调用一次观察者的 onNext 方法，将整个列表传递给它，方法是将 toList 操作符应用于 Observable。
例如，以下相当无用的代码将一个整数列表转换为 Observable，然后将该 Observable 转换为一个发射原始列表作为单个项目的 Observable
numbers = Observable.from([1, 2, 3, 4, 5, 6, 7, 8, 9]);
numbers.toList().subscribe(
{ println(it); }, // onNext
{ println("Error: " + it.getMessage()); }, // onError
{ println("Sequence complete"); } // onCompleted
);[1, 2, 3, 4, 5, 6, 7, 8, 9] Sequence complete
如果源 Observable 在发射任何项目之前调用 onCompleted，则由 toList 返回的 Observable 将在调用 onCompleted 之前发射一个空列表。如果源 Observable 调用 onError，则由 toList 返回的 Observable 将立即调用其观察者的 onError 方法。
toList 默认情况下不会在任何特定的 Scheduler 上操作。
toList()
toMap 操作符将源 Observable 发射的项目收集到一个映射中（默认情况下为 HashMap，但您可以选择提供一个生成另一个 Map 种类的工厂函数），然后发射该映射。您提供一个函数来为每个发射的项目生成键。您也可以选择提供一个函数，将发射的项目转换为要存储在映射中的值（默认情况下，项目本身就是此值）。
toMap 默认情况下不会在任何特定的 Scheduler 上操作。
toMap(Func1)toMap(Func1,Func1)toMap(Func1,Func1,Func0)
toMultiMap 操作符类似于 toMap，不同之处在于它生成的映射也是一个 ArrayList（默认情况下；或者您可以将一个可选的工厂方法作为第四个参数传递，通过该方法您可以生成您喜欢的集合类型）。
toMultiMap 默认情况下不会在任何特定的 Scheduler 上操作。
toMultiMap(Func1)toMultiMap(Func1,Func1)toMultiMap(Func1,Func1,Func0)toMultiMap(Func1,Func1,Func0,Func1)
toSortedList 操作符的行为与 toList 非常相似，不同之处在于它会对结果列表进行排序。默认情况下，它会使用 Comparable 接口按升序对列表进行自然排序。如果源 Observable 发射的任何项目都不支持 Comparable，而该项目的类型与源 Observable 发射的每个项目的类型相同，则 toSortedList 将抛出一个异常。但是，您可以通过将一个函数也传递给 toSortedList 来更改此默认行为，该函数将两个项目作为参数，并返回一个数字；然后，toSortedList 将使用该函数而不是 Comparable 来对项目进行排序。
例如，以下代码将一个无序整数列表转换为 Observable，然后将该 Observable 转换为一个发射原始列表作为单个项目的 Observable，但已排序。
numbers = Observable.from([8, 6, 4, 2, 1, 3, 5, 7, 9]);
numbers.toSortedList().subscribe(
{ println(it); }, // onNext
{ println("Error: " + it.getMessage()); }, // onError
{ println("Sequence complete"); } // onCompleted
);[1, 2, 3, 4, 5, 6, 7, 8, 9] Sequence complete
这是一个提供自己的排序函数的示例：在本例中，一个根据数字与数字 5 的接近程度对数字进行排序的函数。
numbers = Observable.from([8, 6, 4, 2, 1, 3, 5, 7, 9]);
numbers.toSortedList({ n, m -> Math.abs(5-n) - Math.abs(5-m) }).subscribe(
{ println(it); }, // onNext
{ println("Error: " + it.getMessage()); }, // onError
{ println("Sequence complete"); } // onCompleted
);[5, 6, 4, 3, 7, 8, 2, 1, 9] Sequence complete
toSortedList 默认情况下不会在任何特定的 Scheduler 上操作。
toSortedList()toSortedList(Func2)
RxGroovy 还具有一个 nest 操作符，它具有一个特定的用途：它将一个源 Observable 转换为一个 Observable，该 Observable 会发射该源 Observable 作为其唯一的项目。
getIterator 操作符适用于 BlockingObservable 子类，因此为了使用它，您必须首先使用 BlockingObservable.from 方法或 Observable.toBlocking 操作符将源 Observable 转换为 BlockingObservable。
此操作符将 Observable 转换为一个 Iterator，您可以使用它迭代源 Observable 发射的项目集。
BlockingObservable.getIterator()
toFuture 操作符适用于 BlockingObservable 子类，因此为了使用它，您必须首先使用 BlockingObservable.from 方法或 Observable.toBlocking 操作符将源 Observable 转换为 BlockingObservable。
此操作符将 Observable 转换为一个 Future，它将返回源 Observable 发射的单个项目。如果源 Observable 发射了多个项目，Future 将收到一个 IllegalArgumentException；如果它在发射任何项目后完成，Future 将收到一个 NoSuchElementException。
如果您想将一个可能发射多个项目的 Observable 转换为 Future，请尝试以下操作：myObservable.toList().toBlocking().toFuture()。
BlockingObservable.toFuture()
toIterable 操作符适用于 BlockingObservable 子类，因此为了使用它，您必须首先使用 BlockingObservable.from 方法或 Observable.toBlocking 操作符将源 Observable 转换为 BlockingObservable。
此操作符将 Observable 转换为一个 Iterable，您可以使用它迭代源 Observable 发射的项目集。
BlockingObservable.toIterable()
通常，发射多个项目的 Observable 会为每个这样的项目调用其观察者的 onNext 方法。您可以更改此行为，指示 Observable 组成一个包含这些多个项目的列表，然后仅调用一次观察者的 onNext 方法，将整个列表传递给它，方法是将 toList 操作符应用于 Observable。
如果源 Observable 在发射任何项目之前调用 onCompleted，则由 toList 返回的 Observable 将在调用 onCompleted 之前发射一个空列表。如果源 Observable 调用 onError，则由 toList 返回的 Observable 将立即调用其观察者的 onError 方法。
toList 默认情况下不会在任何特定的 Scheduler 上操作。
toList()
toMap 操作符将源 Observable 发射的项目收集到一个映射中（默认情况下为 HashMap，但您可以选择提供一个生成另一个 Map 种类的工厂函数），然后发射该映射。您提供一个函数来为每个发射的项目生成键。您也可以选择提供一个函数，将发射的项目转换为要存储在映射中的值（默认情况下，项目本身就是此值）。
toMap 默认情况下不会在任何特定的 Scheduler 上操作。
toMap(Func1)toMap(Func1,Func1)toMap(Func1,Func1,Func0)
toMultiMap 操作符类似于 toMap，不同之处在于它生成的映射也是一个 ArrayList（默认情况下；或者您可以将一个可选的工厂方法作为第四个参数传递，通过该方法您可以生成您喜欢的集合类型）。
toMultiMap 默认情况下不会在任何特定的 Scheduler 上操作。
toMultiMap(Func1)toMultiMap(Func1,Func1)toMultiMap(Func1,Func1,Func0)toMultiMap(Func1,Func1,Func0,Func1)
toSortedList 操作符的行为与 toList 非常相似，不同之处在于它会对结果列表进行排序。默认情况下，它会使用 Comparable 接口按升序对列表进行自然排序。如果源 Observable 发射的任何项目都不支持 Comparable，而该项目的类型与源 Observable 发射的每个项目的类型相同，则 toSortedList 将抛出一个异常。但是，您可以通过将一个函数也传递给 toSortedList 来更改此默认行为，该函数将两个项目作为参数，并返回一个数字；然后，toSortedList 将使用该函数而不是 Comparable 来对项目进行排序。
toSortedList 默认情况下不会在任何特定的 Scheduler 上操作。
toSortedList()toSortedList(Func2)
RxJava 还具有一个 nest 操作符，它具有一个特定的用途：它将一个源 Observable 转换为一个 Observable，该 Observable 会发射该源 Observable 作为其唯一的项目。
通常，发射多个项目的 Observable 会为每个这样的项目调用其观察者的 onNext 方法。您可以更改此行为，指示 Observable 组成一个包含这些多个项目的数组，然后仅调用一次观察者的 onNext 方法，将整个数组传递给它，方法是将 toArray 操作符应用于 Observable。
var source = Rx.Observable.timer(0, 1000)
.take(5)
.toArray();
var subscription = source.subscribe(
function (x) { console.log('Next: ' + x); },
function (err) { console.log('Error: ' + err); },
function () { console.log('Completed'); });Next: [0,1,2,3,4] Completed
toArray 位于以下每个发行版中
rx.jsrx.all.jsrx.all.compat.jsrx.compat.jsrx.lite.jsrx.lite.compat.js
toMap 操作符将源 Observable 发射的项目收集到一个 Map 中，然后发射该映射。您提供一个函数来为每个发射的项目生成键。您也可以选择提供一个函数，将发射的项目转换为要存储在映射中的值（默认情况下，项目本身就是此值）。
var source = Rx.Observable.timer(0, 1000)
.take(5)
.toMap(function (x) { return x * 2; }, function (x) { return x * 4; });
var subscription = source.subscribe(
function (x) {
var arr = [];
x.forEach(function (value, key) { arr.push(value, key); })
console.log('Next: ' + arr);
},
function (err) { console.log('Error: ' + err); },
function () { console.log('Completed'); });Next: [0,0,2,4,4,8,6,12,8,16] Completed
toMap 位于以下每个发行版中
rx.all.jsrx.all.compat.jsrx.aggregates.js
通常，发射多个项目的 Observable 会为每个这样的项目调用其观察者的 onNext 方法。您可以更改此行为，指示 Observable 组成一个包含这些多个项目的 Set，然后仅调用一次观察者的 onNext 方法，将整个 Set 传递给它，方法是将 toSet 操作符应用于 Observable。
请注意，这仅在 ES6 环境或已填充的环境中有效。
var source = Rx.Observable.timer(0, 1000)
.take(5)
.toSet();
var subscription = source.subscribe(
function (x) {
var arr = [];
x.forEach(function (i) { arr.push(i); })
console.log('Next: ' + arr);
},
function (err) { console.log('Error: ' + err); },
function () { console.log('Completed'); });Next: [0,1,2,3,4] Completed
toSet 位于以下每个发行版中
rx.all.jsrx.all.compat.jsrx.aggregates.js待定
RxPHP 将此操作符实现为 toArray。
创建一个包含单个元素的可观察序列，该元素是一个包含源序列所有元素的数组。
//from https://github.com/ReactiveX/RxPHP/blob/master/demo/toArray/toArray.php
$source = \Rx\Observable::fromArray([1, 2, 3, 4]);
$observer = $createStdoutObserver();
$subscription = $source->toArray()
->subscribe(new CallbackObserver(
function ($array) use ($observer) {
$observer->onNext(json_encode($array));
},
[$observer, "onError"],
[$observer, "onCompleted"]
));
Next value: [1,2,3,4]
Complete!