4

を実行する方法filter、mapおよび複数のスレッドflatMapを使用する場合:Observable

  def withDelay[T](delay: Duration)(t: => T) = {
    Thread.sleep(delay.toMillis)
    t
  }

  Observable
    .interval(500 millisecond)
    .filter(x => {
      withDelay(1 second) { x % 2 == 0 }
    })
    .map(x => {
      withDelay(1 second) { x * x }
    }).subscribe(println(_))

目標は、複数のスレッドを使用してフィルター操作と変換操作を同時に実行することです。

4

3 に答える 3

0

各操作で Async.toAsync() を使用できます。

パッケージ rxjava-async にあります

ドキュメンテーション

于 2016-07-20T20:24:15.673 に答える
-1

オペレーターが設定された後に定義された特定のスレッドで次のすべてのオペレーターを実行する、observeOnオペレーターを使用する必要があります。

       /**
 * Once that you set in your pipeline the observerOn all the next steps of your pipeline will be executed in another thread.
 * Shall print
 * First step main
 * Second step RxNewThreadScheduler-2
 * Third step RxNewThreadScheduler-1
 */
@Test
public void testObservableObserverOn() throws InterruptedException {
    Subscription subscription = Observable.just(1)
            .doOnNext(number -> System.out.println("First step " + Thread.currentThread()
                    .getName()))
            .observeOn(Schedulers.newThread())
            .doOnNext(number -> System.out.println("Second step " + Thread.currentThread()
                    .getName()))
            .observeOn(Schedulers.newThread())
            .doOnNext(number -> System.out.println( "Third step " + Thread.currentThread()
                    .getName()))
            .subscribe();
    new TestSubscriber((Observer) subscription)
            .awaitTerminalEvent(100, TimeUnit.MILLISECONDS);
}

その他の非同期の例はこちらhttps://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/scheduler/ObservableAsynchronous.java

于 2016-07-21T08:03:20.177 に答える