订阅 Observable 什么都不做

Question

我试图了解 Observables 是如何执行的，但似乎无法让这个简单的代码工作。

public class RxJavaExample {
    public static void main(String[] args) {
        Observable<String> hello = Observable.fromCallable(() -> 
            getHello()).subscribeOn(Schedulers.newThread());

        hello.subscribe();

        System.out.println("End of main!");
    }

    public static String getHello() {
        System.out.println("Hello called in " + 
            Thread.currentThread().getName());
        return "Hello";
    }
}

不应该hello.subscribe()执行getHello()吗？

Answer 1

这是因为您的主线程在后台线程到达getHello 之前完成。尝试在您的main 方法中添加Thread.sleep(5000)，然后退出。

或者，等到您订阅的onCompleted 被调用。

编辑：程序终止的原因是因为 RxJava 产生了daemon 线程。在寻找好的来源时，我还发现了this问题，它可能也回答了它。

Answer 2

@sfiss 是对的，这正如你所期望的那样工作：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class RxJavaExample {
  public static void main(String[] args) throws InterruptedException {
    ExecutorService exec = Executors.newCachedThreadPool();
    Observable<String> hello = Observable.fromCallable(() -> getHello())
        .subscribeOn(Schedulers.from(exec));

    hello.subscribe();

    System.out.println("End of main!");

    exec.shutdown();
    exec.awaitTermination(10, TimeUnit.SECONDS);
  }

  public static String getHello() {
    System.out.println("Hello called in " + Thread.currentThread().getName());
    return "Hello";
  }
}

输出如下：

End of main!
Hello called in pool-1-thread-1

Answer 3

你可能对线程和 Observable 感到困惑，

我过去使用 Observables 的方式是用于 Minecraft 插件上的计时器，我有一个每分钟触发的事件。

public class TimerHandler extends Observable implements Runnable{

    @Override
    public void run() {
        this.setChanged();
        this.notifyObservers();
    }
}

所以这每分钟触发一次，然后将事件添加到计时器队列中，您只需订阅 observable 意味着订阅的调用每分钟触发一次。

public class PlotTimer implements Observer {

    @Override
    public void update(Observable o, Object arg) {
        ......

要订阅我打电话给以下

getServer().getScheduler().scheduleAsyncRepeatingTask(this,timerHandler,1200,1200);
timerHandler.addObserver(new PayDayTimer());
timerHandler.addObserver(new ProfileTimer());
timerHandler.addObserver(new PlotTimer());