等价于 RxJava

Question

我们可以在 C# Rx 中异步执行一些代码，如下所示，使用 Observable.Start()。我想知道 RxJava 中的等价物是什么。

void Main()
{
      AddTwoNumbersAsync (5,4)
      .Subscribe(x=>Console.WriteLine(x));

}
IObservable<int> AddTwoNumbersAsync(int a, int b)
{
      return Observable.Start(() => AddTwoNumbers(a, b));
}
int AddTwoNumbers(int a, int b)
{
  return a + b;
}

Answer 1

您可以将操作推迟到订阅，并确保订阅发生在另一个线程上：

Observable<Integer> sumDeferred = Observable.defer(new Func0<Observable<Integer>>() {
        @Override
        public Observable<Integer> call() {
            return Observable.just(addTwoNumbers(5, 4));
        }
    }).subscribeOn(Schedulers.io());
sumDeferred.subscribe(...);

Answer 2

我会使用 rxJava 的 Flowable 对象。

public static void main(String[] args) throws Exception {
    Flowable.fromCallable(() -> addTwoNumbersAsync(5, 4))
            .subscribe(result -> System.out.println(result));
    Thread.sleep(1000);
}
    
private static int addTwoNumbersAsync(int a, int b) {
    return a + b;
}

方法调用和系统打印将在 rxJava 线程中，而不是在主线程中。您可以通过在 .subscribe(...) 之前添加 .subscribeOn(Schedulers.computation()) 来指定 Flowable 将在其上运行的线程池。例如。

你也可以制作一个方法，returns Flowable，更接近你原来的例子。

public static void main(String[] args) throws Exception {
    addTwoNumbersAsync(5,4)
            .subscribe(result -> System.out.println(result));
    Thread.sleep(1000);
}
    
private static Flowable<Integer> addTwoNumbersAsync(int a, int b) {
    return Flowable.fromCallable(() -> a+b);
}

等价于 RxJava

Equivalent in RxJava

c#

system.reactive

reactive-programming

rx-java