В чем разница между concatMap и flatMap в RxJava

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

Плоская карта использует оператор слияния, в то время как concatMap использует оператор concat.

Как вы видите, выходная последовательность concatMap упорядочена - все элементы, излучаемые первым Observable, излучаются перед любым из элементов, излучаемых вторым Observable,
пока выходная последовательность flatMap объединяется - элементы, излучаемые объединенным Observable, могут появляются в любом порядке, независимо от того, из какого источника Observable они пришли.

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

@Test
public void flatMapVsConcatMap() throws Exception {
    System.out.println("******** Using flatMap() *********");
    Observable.range(1, 15)
            .flatMap(item -> Observable.just(item).delay(1, TimeUnit.MILLISECONDS))
            .subscribe(x -> System.out.print(x + " "));

    Thread.sleep(100);

    System.out.println("\n******** Using concatMap() *********");
    Observable.range(1, 15)
            .concatMap(item -> Observable.just(item).delay(1, TimeUnit.MILLISECONDS))
            .subscribe(x -> System.out.print(x + " "));

    Thread.sleep(100);
}

******** Использование flatMap () *********

1 2 3 4 5 6 7 9 8 11 13 15 10 12 14

******** Использование concatMap () *********

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

Как видно из выходных данных, результаты для flatMap неупорядочены, а для concatMap - нет.

Одно очень важное отличие: concatMap ожидает завершения текущей излучаемой наблюдаемой, а flatMap - нет. flatMap пытается запустить как можно больше. Проще говоря - вы не можете объединить что-то бесконечное. Просто убедитесь, что наблюдаемые, которые вы испускаете в concatMap, могут завершиться, иначе весь поток застрянет в ожидании завершения текущего наблюдаемого, чтобы объединить следующий.

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

FlatMap берет выбросы из наблюдаемого источника, затем создает новый наблюдаемый и объединяет его с исходной цепочкой, а concatMap объединяет его с исходной цепочкой.

Основное отличие состоит в том, что concatMap () будет последовательно объединять каждый сопоставленный Observable и запускать его по одному. Он перейдет к следующему Observable только тогда, когда текущий вызовет onComplete ().

Вот пример flatMap:

private void flatMapVsConcatMap() throws InterruptedException {
    Observable.just(5, 2, 4, 1)
            .flatMap(
                    second ->
                            Observable.just("Emit delayed with " + second + " second")
                                    .delay(second, TimeUnit.SECONDS)
            )
            .subscribe(
                    System.out::println,
                    Throwable::printStackTrace
            );

    Thread.sleep(15_000);
}

В результате получится:

Излучение задерживается на 1 секунду
Излучение задерживается на 2 секунды
Излучение задерживается на 4 секунды
Излучение задерживается на 5 секунд

Вот пример concatMap:

private void flatMapVsConcatMap() throws InterruptedException {
    Observable.just(5, 2, 4, 1)
            .concatMap(
                    second ->
                            Observable.just("Emit delayed with " + second + " second")
                                    .delay(second, TimeUnit.SECONDS)
            )
            .subscribe(
                    System.out::println,
                    Throwable::printStackTrace
            );

    Thread.sleep(15_000);
}

В результате получится:

Излучение задерживается на 5 секунд
Излучение задерживается на 2 секунды
Излучение задерживается на 4 секунды
Излучение задерживается на 1 секунду

Обратите внимание на использование Thread.sleep (), поскольку delay () по умолчанию работает в планировщике вычислений.

Во-первых, flatMap - это то же самое, что mergeMap в Rxjs. Так что это на одну путаницу меньше. Итак, есть две наблюдаемые ..

1) o1: простой список предметов из (['Китти', 'Дональд', 'Бэтмен'])

2) process_o1 (): process_o1 () - это функция, которая принимает в качестве одного параметра «элемент» и что-то делает с ним, а также возвращает Observable, который по завершении выдает «выполнено с помощью [элемент]».

o1.pipe(mergeMap(item => process_o1(item))).subscribe(data => {
console.log(data);
});

Здесь мы увидим: - С Кити покончено.

с Дональдом покончено.

сделано с Бэтменом.

без всякой гарантии, что Китти появится раньше Дональда, а Дональд появится перед Бэтменом. Это потому, что, как только внешний наблюдаемый объект испускает элемент, подписывается внутренний наблюдаемый объект.

=== Но в случае concatMap: -

o1.pipe(concatMap(item => process_o1(item))).subscribe(data => {
console.log(data);
});

У нас есть гарантия следующей последовательности: -

с Кити покончено.

с Дональдом покончено.

сделано с Бэтменом.

Поскольку с оператором concatMap внутренний Observable не подписывается до возврата предыдущего внутреннего Observable.

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

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

flatMap против concatMap

flatMap - объединить - если новый элемент испускается, он имеет приоритет

concatMap - объединить - добавить в конец - испустить полную последовательность и только после этого (предыдущая была завершена) может испустить следующую последовательность

[карта против плоской карты]