订阅可观察的什么都不做

Question

我试图理解可观察对象是如何执行的，但似乎无法让这个简单的代码正常工作。

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";
    }
}

不应该是你好。订阅（）执行getHello（）？

Answer 1

你可能正在混淆线程和可观察对象，

我过去使用可观察的方式是在Minecraft插件上设置计时器，我有一个每分钟触发一次的事件。

public class TimerHandler extends Observable implements Runnable{

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

所以这会每分钟触发一次，然后向计时器队列添加事件，您只需订阅可观察的，这意味着订阅的调用每分钟都会触发一次。

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());

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

这是因为主线程在后台线程到达getHello之前完成。尝试添加线程。退出之前，请在主方法中Hibernate（5000）。

或者，等待您的订阅的未完成的调用。

编辑：程序终止的原因是RxJava生成守护进程线程。在寻找一个好的来源时，我也发现了这个问题，这可能也回答了这个问题。