В чём основная концепция использования RxJava?

Question

[Денис]

Изучая RXJava я с толкнулся с трудностью в понимании того, как было задумано его использование.
При первом знакомстве у меня сложилось мнение, о том, что это реализация патерна "Слушатель" с полезными фичами. То есть, есть объект, который генерирует события, а подписанные на объект слушатели эти события получают, при этом объект генерирующий события может существовать от начала работы приложения до самого конца.
Но потом заметил, что во всех примерах использования библиотеки, всегда есть готовый набор данных, который передаётся Observable при его создании(за исключением случаев, где Observable должен генерировать события в повторяющихся промежутках времени) и никогда не дополняется и не заменяется для существующего Observable.

Отсюда у меня сложилось впечатление, что Observable надо воспринимать, как одноразовый объект.
То есть, у нас есть набор данных, мы создаём Observable с этим набором и он существует пока набор не устареет, как только набор устарел, мы отписываем всех подписчиков нашего Observable и создаём новый Observable с новым набором данных, на который подписываются слушатели заново.

Некоторые статьи рассказывают о том, что можно с помощью RXJava следить за набором данных на клавиатуре, примеры они конечно же не приводят, но я нашёл вот такой пример:
https://github.com/shekhargulati/rxjava-examples/b...
Здесь используется "вечный" цикл для отслеживания нажатий на клавиатуре и мне кажется это не лучший способ траты ресурсов устройства.

Растолкуйте мне пожалуйста, как же предполагается использовать RXJava?

P.S. В своём android проекте я хотел создать Observable в сервисе, который занимается некоторой работой с сетью и подписывать на него activitys/fragments, что бы они реагировали на изменения извне. Однако, если Observable придётся на каждый запрос пересоздавать, то возникает необходимость постоянно переподписывать слушателей, что не очень удобно, хотя и возможно.

Answer 1

[Сергей]

RxJava - это библиотека для реализации реактивного стиля в java
реактивный стиль вещь похожая на более известный функциональный стиль программирования.
выражается он в следующем
есть поток данных, мы применяем к нему некие преобразования и получаем новый поток.
например
(поток запросов в сеть в виде адресов страниц)-> операторЗапросаВСеть -> ОператорПарсинга-> ОператорФормированияДатаОбъектов ->(отображение на экране)
Популярность в андроиде, думаю, приобрел за счет простоты работы с многопоточностью. По крайней мере первые примеры применения в андроиде rx обычно про это.)
вообще, это немного другой способ мышления. отличный от процедурного и ооп.
есть источник, есть получатель и много преобразований.
источник это observable, получатель это subscriber/observer . преобразования это разного рода операторы или их цепочки.
В этом случае мы понимаем приложение как нити потоков данных. такая схема бывает проще чем схема из объектов. иногда.
конечно, мы можем применять и гибриды из этих парадигм.
вообще, это стоит попробовать. возможно она очень вам близка по стилю вашего мышления.

Comments

[Денис]

@red-barbarian
Хорошо, принцип я понял.
Я хотел использовать библиотеку скорее как генератор событий, с нагрузкой в виде данных.
А не подскажите, есть ли в нёй инструменты, которые позволяют использовать так как я хочу?
Я попробовал сделать унаследованный от observable класс, но там фактически всю логику обработки подписчиков надо писать самостоятельно.
Если верить ответу Денис Загаевский, то бесконечные Observable используются, но специфически написанные Observable или есть способ быстро писать такие Observable самостоятельно?

[Денис Загаевский]

Денис, не надо ничего наследовать. Observable.create*, там куча методов. Для ваших горячих потоков вообще достаточно будет Publish/BehaviorSubject. Не поленитесь сначала почитать, а не бросаться в пекло.

[Сергей]

Денис, как говорил Денис Загаевский, можно не наследовать. Есть готовые Subject. Ну или если есть желание

пример

RxJava: convert a listener into an Observable

[Денис Загаевский]

Сергей, ну нет же. Это самый сложный способ, нужно напрягаться и соблюдать контракты. Из сабскрайбера вообще не рекомендуется создавать. Есть простейший способ - создать Observable из Emitter'a.

[Сергей]

Денис Загаевский, Да согласен. Сразу не заметил. Вероятно статья старая. Поэтому так сделано. Я так понимаю ObservableEmitter во второй.
тогда такой пример

spoiler

package fs.rxtest;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Cancellable;

public class Main {
    public interface Listener{
        void onNewBoolean(Boolean bool);
    }

    Listener listener;

    void registeredListener(Listener listener){
        System.out.println("register");
        this.listener = listener;
    }
    void unregisteredListener(){
        System.out.println("unregister");
        this.listener = null;
    }


    public static void main(String[] args) {
        Main main = new Main();

        Observable<Boolean> booleanObservable = Observable.create(
                new ObservableOnSubscribe<Boolean>() {
                    @Override
                    public void subscribe(ObservableEmitter<Boolean> e) throws Exception {
                        main.registeredListener(new Listener() {
                            @Override
                            public void onNewBoolean(Boolean bool) {
                                if(!e.isDisposed()) {
                                    e.onNext(bool);
                                }
                            }
                        });
                        e.setCancellable(new Cancellable() {
                            @Override
                            public void cancel() throws Exception {
                                main.unregisteredListener();
                            }
                        });

                    }
                }
        );



        Disposable subscribe = booleanObservable.subscribe(System.out::println);
        main.setValue(true);
        main.setValue(true);
        main.setValue(true);
        main.setValue(true);
        subscribe.dispose();
        main.setValue(true);
        Disposable subscribe1 = booleanObservable.subscribe(b -> System.out.println("2:"));
        main.setValue(false);


    }

    private void setValue(boolean b) {
        if (listener!=null)
        listener.onNewBoolean(b);
    }
}

[Dmtm]

а что в RxJava происходит при пересоздании активности при повороте?

[Сергей]

Dmtm, по умолчанию rxjava в одном потоке. т.е. если activity создал observable и подписался на него, то при пересоздании он уничтожится вместе с активити.
rxjava помогает в другом случае: например, запросы в сеть нельзя делать в ui-потоке. в этом случае помогают операторы subscribeOn и т.п.
для случая пересоздания активити есть несколько типовых решений, такие как: retained fragment
Moxy
ViewModel из Android Architecture Components
и т.п
и при этом нужно понимать, что настоятельно рекомендуется выделять более-менее крупные бизнес и прочие процессы в отдельные потоки (поток) (не ui-поток)

[Dmtm]

Сергей, спасибо, т.е. как и раньше я правильно понял - rxjava на андроиде бесполезна поскольку не решает проблему с жизненным циклом, а библиотек которые отправляют запросы в фоне - сотни

[Сергей]

Dmtm, она удобна и полезна. И к нужному месту. Просто нужно пару небольшие проектиков написать, что бы понять это.
Жизненный цикл активити не такая уж печалька в андроиде по сравнению как сопуствующие утечки памяти.

[Dmtm]

Сергей, rxjava выглядит как очередная обертка над listener, с тем же проблемами подписки-отписки и доставки результата конкретному потребителю, который внезапно раз и пересоздался,
есть решение гораздо лучше,
запросы изменяют слой данных и EventBus для уведомлений об изменении в слое данных,
в этом случае связь между запрос-потребитель отсутствует даже на уровне WeakReference

[Сергей]

Dmtm, можно много рассуждать, но лучше попробовать. Хотя бы по двум причинам
Быть в курсе технологии которая довольно популярна
Уметь писать реактивные программы и разбираться в них.
Про обвертку я писал свое мнение. Это другая парадигма.

Answer 2

[Денис Загаевский]

Концепция Rx заключается в основном в том, что существуют потоки данных внутри приложения. Эти потоки где-то начинаются (create) и где-то заканчиваются(subscribe).

Про наборы данных - не знаю, скорее нет. Например, может быть бесконечный Observable, следящий за БД. Когда вы к нему подписываетесь, он эмитит текущее состояние БД, когда оно изменяется - эмитит ещё. Переподписка для этого не нужна.

То, что вы говорите про Observable, на который вы подписываетесь и работа идёт вне зависимости от этой подписки, это называется "горячий" Observable. Это можно сделать, но немного непонятно, зачем именно для сети нужно такое поведение. Зачем нужен выполняющийся запрос, если на его результат, возможно, никто больше не подпишется? В такой ситуации полезнее "холодный" Observable, который начинает запрос при подписке и отменяет его при отписке.

Далее, насчёт ввода с клавиатуры. Смотрите RxAndroid, и, в частности, RxTextView. В этом случае абсолютно то же самое - пока никто не слушает, следить за клавиатурой не нужно. Когда появился подписчик - начинаем слушать. И в случае именно андроида нет возможности установить несколько листенеров на одну вьюху, потому не удивляйтесь, когда дважды вызвав RxView.clicks(button).subscribe(Log::d) на одной кнопке, обнаружите, что эмиттит только второй Observable, а первый замолчал навсегда.