rxjava android что это

Исследуем RxJava 2 для Android

Меня зовут Аркадий, я Android-разработчик в Badoo. В последнее время в нашем блоге много постов про Go, PHP, JS, QA, и я решил разбавить их темами по мобильной разработке. Как раз занимался портированием одного Android-проекта с RxJava 1 на RxJava 2 и читал всё, что можно найти на эту тему в интернете. В частности, доклад Джейка Вортона с конференции GOTO Copenhagen 2016. Мне показалось, что это достойный кандидат на перевод – думаю, многие Android-разработчики задумываются о переходе на RxJava 2, и им интересно, что изменилось по сравнению с первой версией.

Джейк сделал достаточно объёмное введение о реактивном программировании, так что знание RxJava 1 не требуется для понимания статьи. Доклад был подготовлен, когда RxJava2 ещё только готовилась к выпуску (на текущий момент уже выпущена версия 2.1.0).

Почему Reactive?

Почему все вокруг вдруг стали говорить о реактивном программировании? Если вы не можете сделать приложение полностью синхронным, то наличие единственного асинхронного ресурса полностью ломает традиционный императивный стиль программирования, к которому мы привыкли. «Ломает» не в смысле «всё перестаёт работать», а в смысле «приводит к увеличению сложности», и в результате вы начинаете терять все преимущества императивного программирования.

Чтобы пояснить, почему я считаю это серьёзной проблемой, приведу пример.

Начнём с простого класса, который может получить для нас объект User с какими-то модификаторами.

Если бы мы жили в синхронном, однопоточном мире, то этот код делал бы именно то, что и ожидается: создание экземпляра, вывод пользователя, изменение каких-то свойств, вывод пользователя.

Проблема возникает тогда, когда мы начинаем прибегать к асинхронности. Допустим, нам нужно отразить изменения свойств на стороне сервера. Для этого последние два метода должны быть асинхронными. Как бы мы тогда изменили код?

Одно из решений — вообще ничего не делать: вы можете предположить, что асинхронный вызов обновления сервера будет успешным, так что можно внести изменения локально. Они будут отражены мгновенно. Как вы понимаете, это не лучшая идея. Сети непредсказуемы, сервер может вернуть ошибку, и тогда придётся как-то откатывать локальное состояние.

Можем сообщить пользователю о проблеме. Можем автоматически попытаться снова. Подобные решения работают, и в этом направлении идут, когда нужно совместить асинхронный код с кодом, выполняющимся в одиночном потоке (в случае Android это UI-поток).

Чем больше вам нужно сделать асинхронных вызовов, тем больше возникает проблем. Например, когда пользователь заполняет форму, он изменяет несколько свойств. Или у вас есть последовательность асинхронных вызовов, когда успешное выполнение одного вызова должно запустить другой асинхронный вызов, которому тоже может сопутствовать успех или неудача.

Есть императивные подходы к решению проблемы. Мы можем проверить состояние перед обращением к методам UI.

В этом примере мы создаём анонимный тип, который однозначно приведёт к краткосрочной утечке памяти, потому что он будет хранить ссылку на нашу Activity до тех пор, пока асинхронный вызов не завершится.

Проблема заключается ещё и в том, что мы не знаем, в каких потоках вызываются эти колбэки. Возможно, они вызываются в фоновом потоке, так что требуется передавать события в основной поток выполнения ( main/UI thread ).

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

Реактивное мышление

Пользователи — тоже нечто вроде асинхронных источников данных. Мы даём им информацию через UI, и они реагируют на неё нажатиями кнопок и внесением данных в поля.

Пока вы не можете обеспечить синхронность всей архитектуры вашего приложения, наличие единственного асинхронного ресурса будет ломать традиционный императивный стиль программирования.

Трудно найти приложение, не использующее сетевые запросы, а они по своей природе асинхронны. У вас есть диск, база данных — асинхронные источники. UI тоже должен рассматриваться исключительно как асинхронный источник. Так что по умолчанию всё в Android функционирует асинхронно. Если будете цепляться за традиционное императивное программирование и методики управления состояниями, то будете вредить сами себе.

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

Было бы прекрасно, если бы получаемый нами сетевой ответ обновлял данные. Ведь когда обновляются данные, автоматически обновляется и UI. Таким образом мы снимаем с себя ответственность за это. Если Android делает что-то асинхронно (например, поворот экрана или рассылка broadcast), то было бы замечательно, если бы это автоматически отразилось на интерфейсе или при этом автоматически запускалась какая-нибудь фоновая задача.

В целом такой подход позволяет нам не писать кучу кода, необходимого для поддержки состояний: вместо того чтобы управлять состояниями, мы просто соединяем компоненты друг с другом.

RxJava

Переходим к RxJava. Эта реактивная библиотека стала наиболее популярной при разработке под Android по большей части потому, что была первым полноценным [реактивным] инструментом для Java. RxJava 2 сохраняет поддержку старой версии Java, что важно для разработки под Android.

Источники

Источник данных делает некую работу, когда вы начинаете или заканчиваете его прослушивать. Представьте сетевой запрос, который не будет отправлен, пока вы не начнёте ожидать ответа. И если вы отписываетесь от источника данных до его завершения, то теоретически он может отменить сетевой запрос.

Источник может работать как синхронно, так и асинхронно. Например, блокирующий сетевой запрос, выполняющийся в фоновом потоке, либо что-то чисто асинхронное вроде обращения к Android и ожидания onActivityResult. Источник может выдать один элемент или несколько элементов. Сетевой запрос вернёт один ответ. Но пока работает ваш UI, поток нажатий на кнопки потенциально бесконечен, даже если вы подписаны на единственную кнопку.

Завершения может и не быть. К примеру, мы моделируем нажатие кнопки в качестве источника данных, который работает до тех пор, пока работает UI. И когда UI исчезает, то вы, вероятно, отписываетесь от этого источника нажатий кнопки, но он не завершает свою работу.

Flowable vs. Observable

В RxJava 2 источники представлены двумя основными типами — Flowable и Observable. Они устроены очень похоже. Оба генерируют от нуля до n элементов. Оба могут завершаться успешно или с ошибкой. Так зачем нам два разных типа для представления одной и той же структуры данных?

Всё сводится к такой штуке, как backpressure. Не углубляясь в подробности, скажу лишь, что backpressure позволяет замедлить работу источника данных. Существующие системы имеют ограниченные ресурсы. И с помощью backpressure мы можем сказать всем, кто шлёт нам данные, чтобы они притормозили, потому что мы не можем обрабатывать информацию с той скоростью, с которой она к нам поступает.

В RxJava 1 была поддержка backpressure, но она была добавлена довольно поздно в процессе развития API. В RxJava 1 каждый тип в системе имеет механизм backpressure. И хотя концепцию backpressure поддерживают все типы, далеко не все источники её реализуют, так что использование этого механизма может привести к падению приложения. Применение backpressure должно проектироваться и учитываться заранее. Именно поэтому в RxJava 2 два разных типа источников. Поэтому теперь вы можете указывать с помощью типа источника, должна ли осуществляться поддержка backpressure.

Допустим, у нас есть источник данных — события касания экрана. Мы не можем его замедлить. Нельзя же сказать пользователю: «Нарисуй-ка половину символа, остановись и подожди, пока я обработаю, а затем дорисуй оставшееся». Мы можем замедлить ввод данных иначе, например, отключив кнопки или отобразив другой UI, но сам источник замедлить не получится.

Возьмём другой пример: у нас есть база данных, содержащая большой набор строк, из которого нам нужно извлекать по несколько за раз. БД может очень эффективно решить эту задачу благодаря такому инструменту, как курсоры. Но для потока событий касания это реализовать невозможно, потому что нельзя замедлить палец пользователя.

Итак, единственная разница между двумя этими типами заключается в том, что один поддерживает backpressure, а другой — нет.

Реактивные потоки

Тип Flowable реализует спецификацию реактивных потоков, что подразумевает поддержку backpressure.

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

Специализированные источники

Создание источников

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

Но наиболее полезны два метода, которые, как мне кажется, больше всего будут использоваться для адаптации методов и действий, которые вы уже выполняете (как синхронно, так и асинхронно).

fromCallable доступен для всех пяти типов:

Так можно моделировать синхронные источники одиночных данных. При императивном подходе мы использовали бы метод, возвращающий значение.

Для Maybe и Completable есть ещё по два дополнительных метода. Они позволяют моделировать источники, не возвращающие значения, – просто исполняемые куски кода, но зато реактивные.

Преобразуем в лямбду.

Можно также отправить больше одной порции данных.

Вызывать onNext можно многократно.

Ещё одно преимущество — теперь мы можем моделировать асинхронные данные. К примеру, если нужно асинхронно выполнить HTTP-запросы, то можно вызывать onNext из колбэка HTTP-запроса.

Создание с помощью метода create работает для всех пяти типов:

Наблюдение за источниками

На самом деле, вам необязательно реализовывать интерфейсы Observer/ Subscriber напрямую, когда вы подписываетесь с помощью метода subscribe. Это не вполне удобно как раз из-за четвёртого метода onSubscribe – в нём надо как-то обрабатывать передаваемый объект Disposable / Subscription в момент подписки.

На Android вы часто будете сталкиваться с тем, что у вас есть CompositeDisposable для Activity или фрагмента, и вы отписываетесь в onDestroy (или ещё где-то).

Метод subscribeWith существует для всех четырёх типов без поддержки backpressure.

Можно провести аналогию с ресурсом, файлом, курсором в БД. Вы не откроете файл, не имея способа каким-то образом его закрыть. Открыв курсор в БД, вы в конце концов закроете его. Всегда сохраняйте Disposable при подписке на Observable, чтобы потом отписаться.

Операторы

Операторы позволяют решать три задачи:

Рассмотрим первые две.
Точно так же, как источники позволяют оборачивать синхронные методы и делать их реактивными, операторы позволяют делать реактивными некие действия. Например, здесь мы применяем метод toUppercase() к строке и получаем новую строку.

В реактивном мире мы берем Observable и с помощью оператора осуществляем эту операцию.

Посмотрим на объект User : мы сделали так, чтобы колбэки вызывались в фоновом потоке, и нам приходилось явно передавать данные в основной поток выполнения. На самом деле, это можно сделать с помощью встроенного оператора, причём куда более явным образом.

Можно сказать: «Хочу в другом потоке наблюдать за генерируемыми этим Observable событиями». При смене потоков выполнения имеет значение очерёдность осуществления этих операций.

Возьмём сетевой запрос. Он будет выполняться синхронно, но мы не хотим, чтобы это происходило в основном потоке выполнения. Можно применить оператор, который изменит поток (в котором произойдёт подписка), и работа будет происходить в заданном потоке. В примере мы подписываемся, указывая Schedulers.io() — это просто пул потоков. Работа будет выполнена в этом пуле, а затем всем подписавшимся будут разосланы уведомления. Здесь subscribeOn — это оператор, который задаёт поток выполнения работы.

Специализация операторов

Если Observable был пустой, то это приведёт к ошибке, потому что Single либо содержит элемент, либо возвращает ошибку.

Всё описанное применимо и к Flowable : здесь есть точно такие же операторы, и они возвращают такие же специализированные типы.

В этой таблице приведены некоторые операторы.

Реактивный подход

Если переписывать наш исходный пример с точки зрения реактивного программирования, то можно подписаться на User и сказать: «Хочу получать уведомления в основном потоке выполнения, а затем хочу передать это в UI и отобразить данного пользователя». Этот код будет автоматически выполняться при каждом изменении User’а, и вам будут автоматически показываться сделанные изменения, так что больше не придётся забивать себе этим голову.

Аналогично, когда мы делаем асинхронный запрос на изменение данных, нам нужно, чтобы это произошло в фоновом потоке выполнения. Мы хотим в основном потоке наблюдать за результатом – будет ли выполнение завершено успешно или с ошибкой. И в случае успешного завершения мы могли бы в колбэке заново активировать текстовое поле.

Эта схема требует размещения в памяти кучи промежуточных объектов. В RxJava 2 их количество уменьшено. Каждый из операторов создаёт на один объект меньше, накладные расходы при подписке ниже. Система работает быстрее, нам нужно реже собирать мусор, и всё это – без всяких компромиссов API.

Заключение

В RxJava 2 реализована следующая идея: берутся изначально асинхронные вещи — работа с сетью, сама Android, база данных, даже UI — и пишется такой код, который реагирует на изменения в этих источниках, а не пытается справиться с изменениями и самостоятельно управлять состояниями.

Если вы используете RxJava 1, то обратите внимание на проект, позволяющий конвертировать типы. С его помощью вы сможете постепенно обновить свои приложения до RxJava 2.

Реактивное программирование позволяет использовать правильную архитектуру — асинхронную. Приветствуйте асинхронность источников, а не пытайтесь самостоятельно управлять всеми состояниями. Комбинируйте источники так, чтобы ваши приложения стали по-настоящему реактивными.

Источник

Многопоточное программирование в Android с использованием RxJava 2

Если вы новичок в общении с RxJava или пытались разобраться в этом, но не довели дело до конца, то ниже вы найдете для себя кое-что новое.

Оригинал статьи написан 29 ноября 2017. Перевод вольный.

Нам в GO-JEK требуется выполнять большое количество асинхронных операций в приложениях и мы не можем позволить себе идти на компромиссы в ущерб скорости работы и плавности пользовательского интерфейса.

Написание сложных многопоточных Android приложений может быть достаточно трудоемким процессом, который время от времени будет вас сильно перегружать из-за необходимости заботиться о большом количестве связанных друг с другом вещей. Это и многие другие причины убедили нас использовать RxJava в разрабатываемых Android приложениях.

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

Почему реактивное программирование?

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

Никаких больше обратных вызовов

Если вы давно разрабатываете под Android, то, должно быть, заметили, как быстро вещи становятся чересчур сложными и неподконтрольными с использованием вложенных обратных вызовов.

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

Простой контроль ошибок

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

Очень простое использование многопоточности

Все мы знаем (и тайно признаем) насколько иногда сложной может быть работа с многопоточностью в Java. Например, выполнение части кода в фоновом потоке и возврат результата обратно в главный поток. Это только звучит просто, но на практике появляется много подводных камней, которые нужно обходить.

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

Преимущества RxJava бесконечны. Мы можем говорить об этом часами и адски утомить вас, но вместо этого давайте копнем глубже и начнем изучать реальную работу с многопоточностью в RxJava.

RxJava НЕ многопоточна по умолчанию

Да, вы прочли всё верно. RxJava по умолчанию не многопоточна в любом случае. Определение, данное для RxJava на официальном сайте, выглядит примерно следующим образом:
«Библиотека для составления асинхронных и основанных на событиях программ с использованием последовательностей (observable sequences) для виртуальной Java машины».

Увидев слово «асинхронных», многие люди ошибочно полагают, что RxJava многопоточна по умолчанию. Да, RxJava поддерживает многопоточность, предлагает множество мощных возможностей для легкой работы с асинхронными операциями, но это не значит что поведение RxJava по умолчанию многопоточно.

Если вы уже немного работали с RxJava, то её знаете базовые конструкции:

Если вы запустите данный пример кода, то ясно увидите, что все действия выполняются в основном потоке приложения (проследите за именами потоков в логе в консоли). Этот пример показывает, что по умолчанию поведение RxJava блокирующее. Всё выполняется в том же потоке, в котором вызван код.

Простой пример

Для того, чтобы начать работать с многопоточностью с применением RxJava необходимо познакомиться с базовыми классами и методами, такими как Schedulers, observeOn/subscribeOn.

Давайте рассмотрим один из самых простых примеров. Допустим, мы хотим получить список объектов Book сетевым запросом и показать его в основном потоке приложения. Довольно общий и понятный пример для начала.

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

Не волнуйтесь, скоро мы перейдем к более продвинутым вещам. Этот пример был предназначен только для того, чтобы вспомнить базовые понятия, прежде чем погрузиться глубже.

Подружимся с планировщиками (Schedulers)

RxJava предоставляет мощный набор планировщиков. Вы не можете получить прямой доступ к потокам или управлять ими. Если вам нужно работать с потоками, то необходимо воспользоваться встроенными планировщиками.

Можете представлять планировщики как потоки или пулы потоков (коллекции потоков) для выполнения разного рода задач.

Говоря проще, если вам нужно выполнить задачу в отдельном потоке — необходимо использовать верный планировщик, который возьмёт поток из своего пула доступных потоков и выполнит в нём задачу.

В RxJava доступны несколько типов планировщиков. Самая сложная часть — выбрать верный планировщик для вашей задачи. Задача никогда не будет выполняться оптимально, если вы не выберете верный планировщик. Давайте разберем каждый планировщик.

Schedulers.io()

Этот планировщик основывается на неограниченном пуле потоков и используется для интенсивной работы с вводом-выводом без использования ЦП, например, доступ к файловой системе, выполнение сетевых вызовов, доступ к базе данных и так далее. Количество потоков в этом планировщике неограничено и может расти по мере необходимости.

Schedulers.computation()

Этот планировщик используется для выполнения работы, высоко нагружающей ЦП, такой как обработка больших объемов данных, изображений и так далее. Планировщик основывается на ограниченном пуле потоков с размером в количество доступных процессоров.
Так как этот планировщик подходит только для интенсивной работы с ЦП — количество его потоков ограничено. Сделано это для того, чтобы потоки не конкурировали за процессорное время и не простаивали.

Schedulers.newThread()

Этот планировщик создает совершенно новый поток при каждом вызове. В данном случае использование пула потоков не принесет никакой выгоды. Потоки очень затратно создавать и уничтожать. Вы должны быть осторожны и не злоупотреблять чрезмерным созданием потоков, так как это может привести в замедлению работы системы и переполнению памяти. Новый поток будет создаваться для обработки каждого элемента, полученного из observable-источника.
В идеале вы должны использовать этот планировщик довольно редко, в основном для выведения в отдельный поток долго работающих частей программы.

Schedulers.single()

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

Schedulers.from(Executor executor)

Допустим, вы хотите ограничить число параллельных сетевых вызовов, которые делает ваше приложение. Можно создать собственный планировщик, который будет работать на базе ограниченного в размерах пула потоков ( Scheduler.from(Executors.newFixedThreadPool(n)) ) и использовать его во всех местах, связанных с сетевыми вызовами.

AndroidSchedulers.mainThread()

Понимание subscribeOn() и observeOn()

Теперь, когда у вас есть представление о типах планировщиков, разберем subscribeOn() и observeOn() в деталях.

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

subscribeOn()

Простыми словами, этот оператор говорит в какой поток наблюдаемый источник (source observable) будет передавать элементы. Вы должны уяснить важность слова «источник». Когда у вас цепь наблюдаемых элементов (observables), источник (source observable) — это всегда корневой элемент или верхняя часть цепи, откуда происходит создание событий.

Также важно понять, что нельзя использовать subscribeOn() несколько раз в одной цепочке вызовов. Можно, конечно, написать ещё раз, но никаких изменений это не повлечет. В примере ниже мы последовательно вызываем три различных планировщика, можете ли вы догадаться, какой планировщик сработает при запуске?

Под капотом

Теперь для вас должна быть понятна концепция работы подписок в RxJava. К настоящему времени у вас должно появиться понимание того, как формируются цепочки наблюдаемых (observable) объектов и как события распространяются, начиная с observable-источника.

observeOn()

Как мы уже видели, subscribeOn() указывает observable-источнику передавать элементы в определенный поток и этот поток будет отвечать за продвижение элементов вплоть до подписчика (Subscriber). Поэтому, по умолчанию, подписчик получает обработанные элементы в этом же потоке.

Но это может быть не то поведение, которого вы ожидаете. Предположим, вы хотите получить некие данные из сети и отобразить их в пользовательском интерфейсе.

Нужно выполнить две вещи:

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

Теперь рассмотрим более сложный пример, в котором observeOn() будет вызываться несколько раз для переключения потоков в процессе обработки данных.

Но что произойдет, если использовать observeOn() несколько раз последовательно? В примере ниже в каком потоке подписчик получит результат?

Под капотом

Резюме

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

Если понимание не пришло сразу, ничего страшного. Прочитайте статью ещё раз, поэкспериментируйте с примерами кода. Здесь достаточно много нюансов для понимания, не торопитесь.

Источник