flatMap () против concatMap () против concatMapEager () — RxJava FAQ

В RxJava 2.x есть три одинаково схожих оператора: flatMap() , concatMap() и concatMapEager() . Все они принимают один и тот же аргумент — функцию от отдельного элемента исходного потока до (под) потока произвольного типа. Другими словами, если у вас есть Flowable<T> вы предоставляете функцию от T до Flowable<R> для произвольного типа R После применения любого из этих операторов вы получите Flowable<R> . Так чем они отличаются?

Пример проекта

Сначала давайте создадим пример приложения. Мы будем использовать клиентскую оболочку Retrofit2 HTTP со встроенными плагинами для RxJava2. Наша задача — использовать API GeoNames , чтобы найти население любого города в мире. Интерфейс выглядит следующим образом:

Реализация этого интерфейса создается автоматически Retrofit, прокрутите вниз, чтобы увидеть исходный код клея. Пока предположим, что у нас есть функция, которая принимает String с именем города и асинхронно возвращает поток из одного элемента с населением этого города. Также предположим, что у нас есть фиксированный поток городов, которые мы хотим посмотреть:

Наша цель — собрать население каждого города.

concatMap() : последовательно обрабатывать

Пример приложения с concatMap() выглядит следующим образом:

Прежде чем мы увидим результат, давайте изучим, что concatMap() под ним. Для каждого вышестоящего события ( города ) он вызывает функцию, которая заменяет это событие (под) потоком. В нашем случае это одноэлементный поток Long ( Flowable<Long> ). Таким образом, со всеми операторами, которые мы сравниваем, мы получаем поток потоков Long ( Flowable<Flowable<Long>> ). Реальная разница возникает, когда мы анализируем, что делает оператор, чтобы сгладить такой вложенный поток.

concatMap() сначала подпишется на самый первый подпоток ( Flowable<Long> представляющий население Варшавы). Под подпиской мы фактически подразумеваем физический HTTP-вызов. Только когда первый подпоток завершается (в нашем случае выдает один Long и сигнализирует о завершении) concatMap() будет продолжаться. Продолжение означает подписку на второй подпоток и ожидание его завершения. Результирующий поток завершается, когда завершается самый последний подпоток. Это приводит к следующему потоку: 1702139, 2138551, 7556900 и 3255944. Это, соответственно, популяции Варшавы, Парижа, Лондона и Мадрида. Порядок вывода полностью предсказуем. Однако это также совершенно последовательно. Никакого параллелизма не происходит вообще, мы делаем второй HTTP-вызов только после завершения первого. Дополнительная сложность RxJava вообще не окупается:

Как вы можете видеть, многопоточность не происходит, запросы являются последовательными, ожидая друг друга. Технически не все из них должны происходить в одном потоке, но они никогда не пересекаются и не используют преимущества параллелизма. Большой плюс — это гарантированный порядок результирующих событий, который не так очевиден, как только мы перейдем к flatMap() …

flatMap() : обработка результатов на лету, не в порядке

flatMap() практически такой же:

И так же, как прежде, чем мы начнем с потока потоков Long ( Flowable<Flowable<Long>> ). Однако вместо того, чтобы подписываться на каждый подпоток один за другим, оператор flatMap() охотно подписывается на все подпотоки одновременно. Это означает, что мы видим несколько HTTP-запросов, инициируемых одновременно в разных потоках:

Когда какой-либо из нижележащих подпотоков испускает какое-либо значение, оно немедленно передается абоненту в нисходящем направлении. Это означает, что теперь мы можем обрабатывать события «на лету» по мере их возникновения. Обратите внимание, что результирующий поток вышел из строя. Первое событие, которое мы получили, это 7556900, которое, как оказалось, является населением Лондона, второе в первоначальном потоке. В отличие от concatMap() , flatMap() не может сохранить порядок, поэтому выдает значения в «случайном» порядке. Ну, не совсем случайно, мы просто получаем значения, как только они становятся доступны. В этом конкретном исполнении HTTP-ответ для Лондона появился первым, но на это нет абсолютно никаких гарантий. Это приводит к интересной проблеме. У нас есть поток различных ценностей населения и первоначальный поток городов. Однако выходной поток может быть произвольной перестановкой событий, и мы не знаем, какое население соответствует какому городу. Мы рассмотрим эту проблему в следующей статье.

concatMapEager() : одновременный, по порядку, но несколько дорогой

concatMapEager() кажется, приносит лучшее из обоих миров: параллелизм и гарантированный порядок выходных событий:

Изучив, что concatMap() и flatMap() , понять concatMapEager() довольно просто. Имея поток потоков concatMapEager() охотно ( duh! ) Подписывается на все подпотоки одновременно, одновременно. Однако этот оператор гарантирует, что результаты первого подпотока распространяются первыми, даже если он не завершен первым. Пример быстро покажет, что это значит:

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

Обратите внимание, что население Лондона, хотя и доступно, должно бездействовать, пока не завершатся Варшава и Париж. Так является ли concatMapEager() лучшим из возможных операторов для параллелизма? Не совсем. Представьте, что у нас есть список из тысяч городов, и для каждого мы получаем одну картинку размером 1 МБ. С concatMap() мы загружаем картинки последовательно, т.е. медленно. С flatMap() изображения загружаются одновременно и обрабатываются по мере их поступления, как можно скорее. Теперь, что насчет concatMapEager() ? В худшем случае мы можем получить concatMapEager() буферизующий 999 изображений, потому что изображение из самого первого города оказывается самым медленным. Несмотря на то, что у нас уже есть 99,9% результатов, мы не можем их обработать, потому что мы применяем строгий порядок.

Какой оператор использовать?

flatMap() должен быть вашим первым оружием выбора. Это позволяет эффективный параллелизм с потоковым поведением. Но будьте готовы получить результаты не по порядку. concatMap() работает хорошо, только если преобразование выполняется так быстро, последовательная обработка не является проблемой. concatMapEager() очень удобен, но следите за потреблением памяти. Также в худшем случае вы можете сидеть без дела, ожидая очень мало ответов.

Приложение: настройка клиента Retrofit2

Интерфейс сервиса GeoNames который мы использовали в этой статье, на самом деле выглядит так:

Реализация метода не по умолчанию автоматически генерируется Retrofit2. Обратите внимание, что populationOf() для простоты возвращает одноэлементный Flowable<Long> . Однако, чтобы полностью охватить природу этого API, в реальной жизни было бы более разумно использовать другие реализации. Прежде всего, класс SearchResult возвращает упорядоченный список результатов (методы getter / setters опущены):

Ведь в мире много Варшав и Лондонов . Мы молча предполагаем, что список будет содержать хотя бы один элемент, и первый из них будет правильным . Более подходящая реализация должна либо возвращать все попадания, либо даже лучше тип Maybe<Long> чтобы не отображать совпадения:

Клеевой код выглядит следующим образом. Первая настройка Джексона для анализа ответа от API:

FAIL_ON_UNKNOWN_PROPERTIES часто то, что вы хотите. В противном случае вы должны отобразить все поля из ответа JSON, и ваш код будет поврежден, когда производитель API вводит новые, иначе обратно совместимые поля. Затем мы OkHttpClient , используемый под Retrofit:

Иногда вы можете пропустить настройку клиента OkHttp, но мы добавили перехватчик логирования. По умолчанию протоколы OkHttp используют java.util.logging поэтому, чтобы использовать подходящую среду ведения журналов, мы должны установить мост в самом начале:

И, наконец, дооснащение себя

Вызов createClient() приведет к динамической реализации интерфейса GeoNames . Мы использовали следующие зависимости: