我有些困难理解RxJava中的subscribeOn/observeOn如何工作。我创建了一个简单的应用程序,其中observable发出太阳系行星名称,进行一些映射和过滤并打印结果。
据我所知,通过subscribeOn
运算符将工作调度到后台线程(并且似乎运作良好)。
使用observeOn
运算符在后台线程上进行观察也可以正常工作。
但是,我不明白如何在调用线程上观察(无论它是主线程还是其他线程)。在Android上很容易通过AndroidSchedulers.mainThread()
运算符实现,但我不知道如何在纯java中实现这一点。
这是我的代码:
public class Main {
public static void main(String[] args) throws InterruptedException {
ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 5, 3000L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
System.out.println("Main thread: " + getCurrentThreadInfo());
Observable<String> stringObservable = Observable.from(Arrays.asList("Merkury", "Wenus", "Ziemia", "Mars", "Jowisz", "Saturn", "Uran", "Neptun", "Pluton"))
.map(in -> {
System.out.println("map on: " + getCurrentThreadInfo());
return in.toUpperCase();
})
.filter(in -> {
System.out.println("filter on: " + getCurrentThreadInfo());
return in.contains("A");
})
.subscribeOn(Schedulers.from(executor));
for (int i = 0; i < 5; i++) {
Thread thread = new Thread("Thread-" + i) {
@Override
public void run() {
stringObservable
.buffer(5)
.subscribe(s -> System.out.println("Result " + s + " on: " + getCurrentThreadInfo()));
}
};
thread.start();
}
}
private static String getCurrentThreadInfo() {
return Thread.currentThread().getName() + "(" + Thread.currentThread().getId() + ")";
}
}
在创建过程中观察到的可观察对象并且工作正在从执行器的三个线程中的一个订阅。这按预期工作。但是如何在动态创建的线程上观察结果?有没有一种方法可以从当前线程创建调度程序?
另外,我发现运行此代码后,它永远不会终止,我不知道为什么? :(