我如何运行' Observable '在一个单独的线程

Question

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

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

我永远不会得到"hello"消息，因为while阻塞了一个线程。我怎么能运行我的Observable在一个单独的线程，以避免阻塞?

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

我认为observeOn可以帮助，但它没有。运行这个:

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使OnNext在一个单独的线程中被调用，而不是while块本身?

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

我当然可以把while括在Future中:

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

但是也许有更多的习惯用法来做到这一点呢?

Answer 1

您需要使用subscribeOn。例如,

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

subscribeOn将调用Scheduler中的subscribe函数，而observeOn将向Scheduler分发消息。