Rx.Observable subscribe 和 forEach 有什么区别

Question

像这样创建 Observable 之后

var source = Rx.Observable.create(function(observer) {...});

订阅

有什么区别

source.subscribe(function(x) {});

和forEach

source.forEach(function(x) {});

Answer 1

在RxJS 5.0 后面的ES7 spec 中（但RxJS 4.0 没有），两者并不相同。

订阅

public subscribe(observerOrNext: Observer | Function, error: Function, complete: Function): Subscription

Observable.subscribe 是您进行大部分真正的 Observable 处理的地方。它返回一个订阅令牌，您可以使用它来取消您的订阅。当您不知道已订阅的事件/序列的持续时间，或者您可能需要在已知持续时间之前停止收听时，这一点很重要。

forEach

public forEach(next: Function, PromiseCtor?: PromiseConstructor): Promise

Observable.forEach 返回一个承诺，当 Observable 完成或出错时，该承诺将解决或拒绝。它旨在阐明您以更“同步”的方式处理有界/有限持续时间的可观察序列的情况，例如整理所有传入值，然后通过处理承诺呈现一次。

实际上，您可以对每个值以及错误和完成事件采取任何一种方式。因此，最显着的功能差异是无法取消承诺。

Answer 2

我只是回顾了最新的可用代码，从技术上讲，foreach 的代码实际上是在 RxScala、RxJS 和 RxJava 中调用 subscribe。好像差别不大。他们现在有一个返回类型，允许用户有一种方法来停止订阅或类似的。

当我在 RxJava 早期版本上工作时，订阅有订阅返回，而 forEach 只是一个 void。由于这些变化，您可能会看到一些不同的答案。

/**
 * Subscribes to the [[Observable]] and receives notifications for each element.
 *
 * Alias to `subscribe(T => Unit)`.
 *
 * $noDefaultScheduler
 *  
 * @param onNext function to execute for each item.
 * @throws java.lang.IllegalArgumentException if `onNext` is null
 * @throws rx.exceptions.OnErrorNotImplementedException if the [[Observable]] tries to call `onError`
 * @since 0.19
 * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
 */
def foreach(onNext: T => Unit): Unit = {
    asJavaObservable.subscribe(onNext)
 }

def subscribe(onNext: T => Unit): Subscription = {
    asJavaObservable.subscribe(scalaFunction1ProducingUnitToAction1(onNext))
}

/**
 *  Subscribes an o to the observable sequence.
 *  @param {Mixed} [oOrOnNext] The object that is to receive notifications or an action to invoke for each element in the observable sequence.
 *  @param {Function} [onError] Action to invoke upon exceptional termination of the observable sequence.
 *  @param {Function} [onCompleted] Action to invoke upon graceful termination of the observable sequence.
 *  @returns {Disposable} A disposable handling the subscriptions and unsubscriptions.
 */
observableProto.subscribe = observableProto.forEach = function (oOrOnNext, onError, onCompleted) {
  return this._subscribe(typeof oOrOnNext === 'object' ?
    oOrOnNext :
    observerCreate(oOrOnNext, onError, onCompleted));
};

/**
 * Subscribes to the {@link Observable} and receives notifications for each element.
 * <p>
 * Alias to {@link #subscribe(Action1)}
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code forEach} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 * 
 * @param onNext
 *            {@link Action1} to execute for each item.
 * @throws IllegalArgumentException
 *             if {@code onNext} is null
 * @throws OnErrorNotImplementedException
 *             if the Observable calls {@code onError}
 * @see <a href="http://reactivex.io/documentation/operators/subscribe.html">ReactiveX operators documentation: Subscribe</a>
 */
public final void forEach(final Action1<? super T> onNext) {
    subscribe(onNext);
}

public final Disposable forEach(Consumer<? super T> onNext) {
    return subscribe(onNext);
}