代码之家 › 专栏 › 技术社区 › LiTTle

RXJava2块订阅与订阅

rx-java2 kotlin java

LiTTle · 技术社区 · 7 年前

我读过关于 blockingSubscribe() 和 subscribe() 但我既写不出来,也找不到一个例子来说明这些差异。这两种方法似乎都是一样的。有人能提供这2个例子,最好是在Java中。

1 回复 | 直到 7 年前

akarnokd 7 年前

blockingSubscribe 阻塞当前线程并处理其中的incomnig事件。您可以通过运行一些异步源来看到这一点:

System.out.println("Before blockingSubscribe");
System.out.println("Before Thread: " + Thread.currentThread());

Observable.interval(1, TimeUnit.SECONDS)
.take(5)
.blockingSubscribe(t -> {
     System.out.println("Thread: " + Thread.currentThread());
     System.out.println("Value:  " + t);
});

System.out.println("After blockingSubscribe");
System.out.println("After Thread: " + Thread.currentThread());

subscribe 不受限制,可在任意螺纹上运行:

System.out.println("Before subscribe");
System.out.println("Before Thread: " + Thread.currentThread());

Observable.timer(1, TimeUnit.SECONDS, Schedulers.io())
.concatWith(Observable.timer(1, TimeUnit.SECONDS, Schedulers.single()))
.subscribe(t -> {
     System.out.println("Thread: " + Thread.currentThread());
     System.out.println("Value:  " + t);
});


System.out.println("After subscribe");
System.out.println("After Thread: " + Thread.currentThread());

// RxJava uses daemon threads, without this, the app would quit immediately
Thread.sleep(3000);

System.out.println("Done");

推荐文章