RxJava2 - Создание элементов с использованием PublishSubject

Question

У меня есть сценарий, где у меня есть

subject1: PublishSubject и subject2:BehaviorSubject.

Сначала я выбрасываю один предмет за subject1, затем я выдаю предмет за subject2, но сразу после этого я также хочу выдать другой предмет на subject1.

fun emittingItems() {
    subject1.onNext(functionA1)
    subject2.onNext(functionB)
    if (something) subject1.onNext(functionA2)
}

Что происходит, когда я получаю предмет в этой последовательности: functionA1, functionA2, functionB.

Почему у меня такое поведение?Как я могу испускать предметы в этой последовательности: functionA1, functionB, functionA2.

Подписка на темы:

val disposable = viewModel.subject1
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(this::someFunction())
disposables.add(disposable)

Answer 1

С помощью observeOn(AndroidSchedulers.mainThread()) вы планируете распространение событий в главном потоке.Само планирование является последовательным, в то время как каждый запланированный Runnable может обрабатывать более одного элемента, добавленного в очередь, используемую для него.

Это своего рода условие гонки, которое наверняка возникнет при вызове emittingItems() наОсновной поток и может возникнуть при вызове его из любого другого потока.

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

Выможет достичь заданного, объединяя оба источника в один поток:

Observable.merge(subject1, subject2)
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(subject);