等待RxJS Observable的onNext中的异步操作

Question

我有正常方式消耗的RxJS序列......

但是,在可观察的'onNext'处理程序中,某些操作将同步完成,但其他操作需要异步回调,需要在处理输入序列中的下一个项之前等待.

......有点困惑如何做到这一点.有任何想法吗？谢谢!

someObservable.subscribe(
    function onNext(item)
    {
        if (item == 'do-something-async-and-wait-for-completion')
        {
            setTimeout(
                function()
                {
                    console.log('okay, we can continue');
                }
                , 5000
            );
        }
        else
        {
            // do something synchronously and keep on going immediately
            console.log('ready to go!!!');
        }
    },
    function onError(error)
    {
        console.log('error');
    },
    function onComplete()
    {
        console.log('complete');
    }
);

Answer 1

您要执行的每个操作都可以建模为可观察的.甚至同步操作也可以这种方式建模.然后,您可以使用map将序列转换为序列序列,然后用于concatAll展平序列.

someObservable
    .map(function (item) {
        if (item === "do-something-async") {
            // create an Observable that will do the async action when it is subscribed
            // return Rx.Observable.timer(5000);

            // or maybe an ajax call?  Use `defer` so that the call does not
            // start until concatAll() actually subscribes.
            return Rx.Observable.defer(function () { return Rx.Observable.ajaxAsObservable(...); });
        }
        else {
            // do something synchronous but model it as an async operation (using Observable.return)
            // Use defer so that the sync operation is not carried out until
            // concatAll() reaches this item.
            return Rx.Observable.defer(function () {
                return Rx.Observable.return(someSyncAction(item));
            });
        }
    })
    .concatAll() // consume each inner observable in sequence
    .subscribe(function (result) {
    }, function (error) {
        console.log("error", error);
    }, function () {
        console.log("complete");
    });

要回复你的一些评论...在某些时候你需要对功能流强制一些期望.在大多数语言中,当处理可能异步的函数时,函数签名是异步的,函数的实际异步与同步性质被隐藏为函数的实现细节.无论您使用的是javaScript promises,Rx observables,c#Tasks,c ++ Futures等,都是如此.函数最终返回promise/observable/task/future/etc,如果函数实际上是同步的,那么它返回的对象是刚刚完成.

话虽如此,因为这是JavaScript,你可以作弊:

var makeObservable = function (func) {
    return Rx.Observable.defer(function () {
        // execute the function and then examine the returned value.
        // if the returned value is *not* an Rx.Observable, then
        // wrap it using Observable.return
        var result = func();
        return result instanceof Rx.Observable ? result: Rx.Observable.return(result);
    });
}

someObservable
    .map(makeObservable)
    .concatAll()
    .subscribe(function (result) {
    }, function (error) {
        console.log("error", error);
    }, function () {
        console.log("complete");
    });

Answer 2

首先，将您的异步操作移出subscribe，这不是针对异步操作进行的。

您可以使用mergeMap（别名flatMap）或 concatMap。（我提两个人，但concatMap实际上是mergeMap与concurrent参数设置为1）Settting不同的并发参数是非常有用的，因为有时你会想限制并发查询的数量，但仍运行几个并发。

source.concatMap(item => {
  if (item == 'do-something-async-and-wait-for-completion') {
    return Rx.Observable.timer(5000)
      .mapTo(item)
      .do(e => console.log('okay, we can continue'));
    } else {
      // do something synchronously and keep on going immediately
      return Rx.Observable.of(item)
        .do(e => console.log('ready to go!!!'));
    }
}).subscribe();

我还将展示如何为通话限价。忠告：仅在实际需要时限制速率，例如调用外部API时，每秒或每分钟仅允许一定数量的请求。否则，最好限制并发操作的数量，并让系统以最大速度移动。

我们从以下代码段开始：

const concurrent;
const delay;
source.mergeMap(item =>
  selector(item, delay)
, concurrent)

接下来，我们需要选择的值concurrent，delay并实施selector。concurrent和delay密切相关。例如，如果我们想每秒运行10个项目，则可以使用concurrent = 10and delay = 1000（毫秒），也可以使用concurrent = 5and delay = 500或concurrent = 4and delay = 400。每秒的项目数始终为concurrent / (delay / 1000)。

现在开始实施selector。我们有两种选择。我们可以为设置一个最小执行时间selector，我们可以为其添加一个恒定的延迟，我们可以在结果可用时立即发出结果，只有在最小延迟过去之后我们才可以发出结果，等等。通过使用timeout运算符添加超时。方便。

设置最短时间，尽早发送结果：

function selector(item, delay) {
   return Rx.Observable.of(item)
     .delay(1000) // replace this with your actual call.
     .merge(Rx.Observable.timer(delay).ignoreElements())
}

设置最短时间，延迟发送结果：

function selector(item, delay) {
   return Rx.Observable.of(item)
     .delay(1000) // replace this with your actual call.
     .zip(Rx.Observable.timer(delay), (item, _))
}

添加时间，尽早发送结果：

function selector(item, delay) {
   return Rx.Observable.of(item)
     .delay(1000) // replace this with your actual call.
     .concat(Rx.Observable.timer(delay).ignoreElements())
}

添加时间，延迟发送结果：

function selector(item, delay) {
   return Rx.Observable.of(item)
     .delay(1000) // replace this with your actual call.
     .delay(delay)
}