multithreading - 如何在单独的线程中运行 `Observable`？

Question

我在Scala项目中使用RxJava，说我有这个简单的Observable:

Observable[String](observer => while (true) observer onNext "hi")
  .subscribe(v => println(v))

println("hello")

我永远不会收到“hello”消息，因为o​​jit_code阻塞了一个线程。如何在单独的线程中运行 while以避免阻塞？

=================================

我以为 Observable可以帮上忙，但没有帮助。运行此:

val s = rx.lang.scala.schedulers.NewThreadScheduler.apply

Observable[String](observer => while (true) observer onNext "hi")
  .observeOn(s).subscribe(v => println(v))

println("hello")

...仍然不打印“hello”。我猜想添加 observeOn会使 observeOn在单独的线程中被调用，而不是 OnNext块本身吗？

=================================

我当然可以将 while包装在 while中:

Observable[String](observer => Future { while (true) observer onNext "hi" })
  .subscribe(v => println(v))

println("test") // "test" gets printed

但是也许有更多的rx习惯用法可以做到这一点？

Answer 1

您需要使用subscribeOn。例如。，

Observable[String](observer => while (true) observer onNext "hi").subscribeOn(s)
  .subscribe(v => println(v))

subscribeOn将在 subscribe中调用 Scheduler函数，而 observeOn将消息发送到 Scheduler。