Entenda a diferença entre concat, concatMap, merge e mergeMap do RxJS

Trabalhando há alguns anos em projetos Angular, que por padrão instala a biblioteca RxJS, acabei me deparando por diversas vezes com o concat, concatMap, merge e mergeMap. Confesso que precisei gastar um tempo para entender as diferenças, mas fiz um resumo para facilitar esse entendimento nesse pequeno texto.

Combinação de observables: concat vs merge

O concat e merge possibilitam a combinação de múltiplos observables. Ambos recebem uma lista de observables e retornam um único observable. Essa lista de observables que passamos como parâmetro para o concat e merge são definidos como "inner observable". Já o observable que o concat e merge retorna é o "outer observable".

concat

O operador concat do RxJS é usado para combinar múltiplos observables em uma sequência ordenada. Ele recebe uma lista de observables (inner observables) e retorna um único Observable (outer observable).

A principal característica do concat é que ele se inscreve em cada observable interno (inner) um após o outro. Ele primeiro se inscreve no primeiro observable e emite seus valores até que ele complete. Somente após a conclusão do primeiro observable, ele se inscreve no próximo, e assim por diante. Esse processo continua até que todos os observables internos tenham sido concluídos.

merge

Diferentemente do concat, que emite valores sequencialmente, o merge, por se inscrever em todos os observables logo no início, ele emite valores de todos os observables internos simultaneamente, à medida que eles chegam.

Exemplo: concat vs merge

Para facilitar o entendimento, considere o código abaixo onde o método getSources retorna uma lista com 3 observables. Cada um tem um delay para atrasar propositalmente a emissão dos seus valores:

import { Injectable } from '@angular/core';
import { delay, Observable, of, tap } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class SourcesService {
  getSources(
    log: (message: string) => void
  ): Observable<number>[] {
    const source1 = of(1, 2, 3).pipe(
      delay(1500),
      tap({
        subscribe: () => log('source1.subscribe'),
        complete: () => log('source1.complete'),
      })
    );

    const source2 = of(4, 5, 6).pipe(
      delay(1000),
      tap({
        subscribe: () => log('source2.subscribe'),
        complete: () => log('source2.complete'),
      })
    );

    const source3 = of(7, 8, 9).pipe(
      delay(500),
      tap({
        subscribe: () => log('source3.subscribe'),
        complete: () => log('source3.complete'),
      })
    );

    return [source1, source2, source3];
  }
}

Na sequência, deixei a implementação de 2 componentes praticamente iguais. A única diferença entre eles é que o primeiro usa o concat e o segundo o merge:

import { Component } from '@angular/core';
import { concat, Observable } from 'rxjs';
import { LogsComponent } from '../logs/logs.component';
import { SourcesService } from '../sources.service';

@Component({
  selector: 'app-concat',
  imports: [LogsComponent],
  templateUrl: './concat.component.html',
})
export class ConcatComponent {
  logs: Array<{ data: Date; msg: string | number }> = [];
  source1: Observable<number>;
  source2: Observable<number>;
  source3: Observable<number>;

  constructor(sourcesService: SourcesService) {
    [this.source1, this.source2, this.source3] =
      sourcesService.getSources(this.log.bind(this));

    concat(
      this.source1,
      this.source2,
      this.source3
    ).subscribe((valor) => this.log(valor));
  }

  log(message: string | number) {
    this.logs.push({
      data: new Date(),
      msg: message,
    });
  }
}

import { Component } from '@angular/core';
import { merge, Observable } from 'rxjs';
import { LogsComponent } from '../logs/logs.component';
import { SourcesService } from '../sources.service';

@Component({
  selector: 'app-merge',
  imports: [LogsComponent],
  templateUrl: './merge.component.html',
})
export class MergeComponent {
  // ...(código omitido, igual ao ConcatComponent)...

  constructor(sourcesService: SourcesService) {
    [this.source1, this.source2, this.source3] =
      sourcesService.getSources(this.log.bind(this));

    merge(
      this.source1,
      this.source2,
      this.source3
    ).subscribe((valor) => this.log(valor));
  }

  // ...(código omitido, igual ao ConcatComponent)...
}

Como resultado (imagem abaixo), podemos observar que no caso do concat, a ordem com que os observables foram passados foi respeitada, ou seja, primeiro ele faz o subscribe no primeiro observable (of(1, 2, 3)), recebe seus valores até sua conclusão e então faz o subscribe no próximo (of(4, 5, 6)), e assim por diante.

Já o merge faz o subscribe em todos os observables que ele recebeu como parâmetro logo no início e vai emitindo os valores conforme eles chegam. Aqui não existe uma priorização, o valor que é emitido primeiro dentro dos "inner observable" é emitido para o outer observable.

Transformação de observables: concatMap vs mergeMap

O concatMap e mergeMap, permitem que cada valor emitido de um observable seja transformado em um outro observable.

concatMap

Para cada valor emitido de um observable A, o concatMap retorna um novo observable B. Esse novo observable B, quando concluído, pega o próximo valor emitido pelo observable A e assim por diante. Assim como o concat, ele respeita a ordem dos valores recebidos do observable A.

mergeMap

Diferente do concatMap, o mergeMap retorna um novo observable B assim que o observable A emite um valor.

Exemplo 1 - concatMap e mergeMap

Considere o observable source1 que emite 3 valores na sequência: 3, 2 e 1. Para cada valor emitido por este observable, vamos criar um novo observable com um delay de 3 segundos, 2 segundos e 1 segundo respectivamente.

const source1: Observable<number> = of(3, 2, 1);
source1
  .pipe(
    tap({
      subscribe: () => this.log('1', 'subscribe'),
      next: (a) => this.log('1', `next(${a})`),
      complete: () => this.log('1', 'complete'),
    }),
    concatMap((a) =>
      of(a).pipe(
        tap({
          subscribe: () => this.log('2', 'subscribe'),
          next: (a) => this.log('2', `next(${a})`),
          complete: () => this.log('2', 'complete'),
        }),
        delay(a * 1000)
      )
    )
  )
  .subscribe((valor) => {
    this.log('', valor);
  });

const source1: Observable<number> = of(3, 2, 1);
source1
  .pipe(
    tap({
      subscribe: () => this.log('1', 'subscribe'),
      next: (a) => this.log('1', `next(${a})`),
      complete: () => this.log('1', 'complete'),
    }),
    mergeMap((a) =>
      of(a).pipe(
        tap({
          subscribe: () => this.log('2', 'subscribe'),
          next: (a) => this.log('2', `next(${a})`),
          complete: () => this.log('2', 'complete'),
        }),
        delay(a * 1000)
      )
    )
  )
  .subscribe((valor) => {
    this.log('', valor);
  });

Observe no resultado abaixo que o observable de origem (source1), para ambos, emitem os valores 3, 2 e 1. Cada valor, independente do concatMap e mergeMap, resulta em um novo observable com um delay igual ao valor emitido * 1000 segundos. Ou seja, o observable of(3, 2, 1), após passar pelo concatMap ou mergeMap resultará em respectivamente:

1. of(3).pipe(delay(3000));
2. of(2).pipe(delay(2000));
3. of(1).pipe(delay(1000));

Como o concatMap processa os observables internos em sequência. Ele espera que um observable interno seja concluído antes de se inscrever no próximo. Isso garante que os valores emitidos pelo observable resultante mantenham a mesma ordem do observable de origem. Por isso, primeiro ele vai esperar 3 segundos para emitir o valor 3, depois 2 segundos para emitir o valor 2 e por ultimo 1 segundo para emitir o valor 1, ou seja, ordem preservada, execução sequencial.

Já o mergeMap processa os observables internos simultaneamente (concorrentemente). Ele se inscreve em cada novo Observable interno assim que o observable de origem emite um valor. A ordem dos valores emitidos pelo observable resultante pode não corresponder à ordem do observable de origem. Neste caso, o observable que possui o menor tempo de delay sera executado primeiro, e assim por diante.

Considerações

As funçõesconcat, concatMap, merge e mergeMap do RxJS são fundamentais para controlar o fluxo de eventos em aplicações reativas. O concat garante a execução sequencial dos observables, respeitando sua ordem, enquanto o merge permite a concorrência, emitindo valores conforme chegam. Já o concatMap e o mergeMap funcionam de maneira semelhante, mas no contexto de mapeamento de valores para novos observables. O concatMap mantém a ordem e aguarda a conclusão de cada observable antes de processar o próximo, enquanto o mergeMap dispara múltiplas subscrições sem aguardar a conclusão da anterior.

A escolha entre esses operadores depende do comportamento desejado: se a ordem e o encadeamento são importantes, concat e concatMap são as melhores opções; se a prioridade for a velocidade e a execução paralela, merge e mergeMap são mais indicados.

Links:

• https://rxjs.dev/api/operators/merge
• https://www.learnrxjs.io/learn-rxjs/operators/combination/merge
• https://rxjs.dev/api/operators/concat
• https://www.learnrxjs.io/learn-rxjs/operators/combination/concat
• https://rxjs.dev/api/index/function/concatMap
• https://rxjs.dev/api/index/function/mergeMap