Android RxJava и цепочки наблюдаемых

Question

Я пытаюсь объединить несколько наблюдаемых вместе и сделать какое-то действие в зависимости от того, что наблюдалось. Но я столкнулся со странным поведением.

class MainActivity : AppCompatActivity() { 

    val TAG: String = MainActivity::class.java.name 

    private lateinit var clicker: TextView 

    override fun onCreate(savedInstanceState: Bundle?) { 
     super.onCreate(savedInstanceState) 
     setContentView(R.layout.activity_main) 
     clicker = findViewById(R.id.clicker) as TextView 
     clicker.setOnClickListener { 
      val i = AtomicInteger() 

      getFirstObservable() 
       .subscribeOn(Schedulers.computation()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .doOnNext { 
        showMessage(i, it) 
       } 
       .flatMap { getSecondObservable() } 
       .doOnNext { 
        showMessage(i, it) 
       } 
       .flatMap { getThirdObservable() } 
       .doOnNext { 
        showMessage(i, it) 
       } 
       .subscribe() 
     } 
    } 

    fun getFirstObservable(): Observable<String> { 
     return Observable.fromCallable { 
      Thread.sleep(2000) 
      "Hello" 
     } 
    } 

    fun getSecondObservable(): Observable<Int> { 
     return Observable.fromCallable { 
      Thread.sleep(2000) 
      3 
     } 
    } 

    fun getThirdObservable(): Observable<String> { 
     return Observable.fromCallable { 
      Thread.sleep(2000) 
      "World!" 
     } 
    } 

    fun showMessage(i: AtomicInteger, obj: Any) { 
     val msg = "Message #${i.incrementAndGet()}: from ${Thread.currentThread().name}: $obj" 
     Log.e(TAG, msg) 
     clicker.text = msg 
     Toast.makeText(this, msg, Toast.LENGTH_SHORT).show() 
    } 
} 

В этом примере журналы будут показаны каждые 2 секунды, но все изменения с видом будет сделано, когда последняя наблюдаемым будет закончена.

12-04 01:11:30.465 19207-19207/com.googlevsky.rxtest E/com.googlevsky.rxtest.MainActivity: Message #1: from main: Hello 
12-04 01:11:32.473 19207-19207/com.googlevsky.rxtest E/com.googlevsky.rxtest.MainActivity: Message #2: from main: 3 
12-04 01:11:34.479 19207-19207/com.googlevsky.rxtest E/com.googlevsky.rxtest.MainActivity: Message #3: from main: World! 

Я думаю, что это поведение AndroidScheduler.mainThread(), потому что когда я удалить эту строку и оберните изменения с видом, как это поведение

Handler(Looper.getMainLooper()).post { 
    clicker.text = msg 
    Toast.makeText(this, msg, Toast.LENGTH_SHORT).show() 
} 

становится правильным. Так может кто-нибудь объяснить это поведение и предложить правильный способ решить эту проблему?

Answer 1

Большая часть вашего кода выполняется в основном потоке, включая спальные. Когда наблюдаемое создается, оно подписывается и наблюдается в текущем потоке, если не указано иное. Когда вы создаете свои второй и третий наблюдаемые, они находятся в основном потоке. Кроме того, поскольку нет фоновой информации о наблюдаемой работе, она выполняется сразу по текущему потоку при подписке. Поэтому вся работа и наблюдения происходят в основном потоке, не возвращаясь к ОС Android. Пользовательский интерфейс заблокирован, ожидая времени на основном потоке. Если вы увеличите время сна, вы можете заставить ANR. Чтобы исправить это, вы можете указать observeOn и subscribeOn для каждого из ваших наблюдаемых, чтобы подтолкнуть работу к потоку вычислений для каждого из них.

getFirstObservable().subscribeOn(Schedulers.computation()) 
        .observeOn(AndroidSchedulers.mainThread()) 
        .doOnNext { 
         showMessage(i, it) 
        } 
        .flatMap { 
         getSecondObservable().subscribeOn(Schedulers.computation()) 
              .observeOn(AndroidSchedulers.mainThread()) 
        } 
        .doOnNext { 
         showMessage(i, it) 
        } 
        .flatMap { 
         getThirdObservable().subscribeOn(Schedulers.computation()) 
              .observeOn(AndroidSchedulers.mainThread()) 
        } 
        .doOnNext { 
         showMessage(i, it) 
        } 
        .doOnNext { 
         showMessage(i, it) 
        } 
        .subscribe()