merge, concat, combineLatest, zip, forkJoin: 여러 스트림을 하나로 결합하는 다섯 가지 방법

1. 개념

RxJS에서 스트림 결합(combination)이란 여러 개의 옵저버블을 입력으로 받아서 하나의 옵저버블로 만들어내는 작업을 말한다. 하나의 스트림만으로는 표현하기 어려운, "여러 비동기 소스를 동시에 다뤄야 하는 상황"을 위해 존재하는 연산자군이다.

• merge: 여러 옵저버블을 구독하고, 각 소스에서 값이 발행되는 즉시 순서 상관없이 하나의 스트림으로 흘려보낸다.
• concat: 여러 옵저버블을 순서대로 구독한다. 앞의 옵저버블이 완료(complete)되어야 다음 옵저버블을 구독한다.
• combineLatest: 모든 소스가 최소 한 번씩 값을 낸 이후, 어느 소스든 새 값을 낼 때마다 각 소스의 "최신 값"을 묶어서 방출한다.
• zip: 각 소스에서 같은 순번의 값끼리 짝을 지어 방출한다. 마치 지퍼(zipper)처럼 인덱스가 맞아야 값이 나간다.
• forkJoin: 모든 소스가 완료될 때까지 기다렸다가, 각 소스의 마지막 값만 한 번에 묶어서 방출한다.

2. 왜 사용하는가

실무에서는 하나의 화면이나 기능이 여러 개의 비동기 작업에 의존하는 경우가 많다. 예를 들어 사용자 목록과 권한 정보를 동시에 조회해야 하거나, 여러 개의 버튼 클릭 이벤트를 하나의 로그 스트림으로 합쳐야 하거나, 여러 입력값이 모두 준비된 시점에 계산을 수행해야 하는 경우다. 이런 요구를 개별적으로 구독하고 콜백을 중첩해서 처리하면 코드가 복잡해지고 타이밍 버그가 생기기 쉽다. 결합 연산자는 이런 다중 소스 처리 로직을 선언적으로 표현할 수 있게 해준다.

3. 동작 원리

merge는 내부적으로 전달받은 모든 옵저버블을 동시에 구독(subscribe)한다. 각 소스에서 값이 도착하면 도착한 순서 그대로 출력 스트림에 흘려보낸다. 모든 소스가 완료되어야 결합된 스트림도 완료된다. 소스 간 순서나 타이밍은 전혀 보장하지 않는다.

concat은 첫 번째 옵저버블만 먼저 구독한다. 그 옵저버블이 완료 신호를 보내야 두 번째 옵저버블을 구독하기 시작한다. 따라서 소스의 순서가 결과의 순서를 그대로 결정하지만, 앞선 옵저버블이 완료되지 않는(무한 스트림 같은) 경우 뒤의 옵저버블은 영원히 실행되지 않는다.

combineLatest는 전달받은 모든 옵저버블을 동시에 구독하고, 각 소스별로 "가장 최근에 받은 값"을 내부에 저장해둔다. 모든 소스가 적어도 한 번씩 값을 낸 시점부터 동작을 시작하며, 이후 어느 소스에서든 새 값이 오면 그 시점의 모든 소스의 최신 값을 배열이나 객체로 묶어서 방출한다.

zip도 모든 소스를 동시에 구독하지만, 값을 내부 큐에 순서대로 쌓아두고 모든 소스에서 같은 인덱스의 값이 도착했을 때만 짝을 지어 방출한다. 한 소스가 다른 소스보다 값을 빨리 내더라도, 짝이 맞을 때까지 대기한다.

forkJoin은 모든 소스를 동시에 구독하되, 각 소스가 완료될 때 마지막으로 발행한 값만 기억한다. 모든 소스가 완료되면 그 마지막 값들을 한 번에 묶어서 방출하고 즉시 완료된다. 소스 중 하나라도 완료되지 않거나 에러가 나면 결과가 나오지 않는다.

4. 예제

merge 예제: 두 개의 버튼 클릭 스트림을 하나로 합치고 싶다면, 먼저 fromEvent(button1, 'click')과 fromEvent(button2, 'click')을 만들고, 그다음 merge(click1$, click2$)로 구독하면 어느 버튼을 누르든 즉시 값이 흘러나온다. 두 스트림의 발생 순서가 뒤섞여도 상관없는 로깅이나 알림 처리에 적합하다.

concat 예제: 초기 데이터 로딩을 순서대로 처리하고 싶다면, 먼저 getUserInfo()가 사용자 정보를 가져오는 옵저버블을 반환하게 하고, 그다음 getUserSettings()가 설정 정보를 가져오는 옵저버블을 반환하게 한다. concat(getUserInfo(), getUserSettings())로 구독하면 사용자 정보 요청이 끝난 뒤에야 설정 정보 요청이 시작된다. 두 요청 사이에 순서 의존성이 있을 때 유용하다.

combineLatest 예제: 사용자가 입력한 이름과 나이를 실시간으로 조합해서 미리보기를 보여주고 싶다면, 먼저 name$과 age$라는 두 개의 입력 스트림을 만들고, 그다음 combineLatest([name$, age$])로 구독한다. 이름이나 나이 중 하나라도 바뀌면 그 시점의 최신 이름과 나이를 함께 받아서 미리보기 텍스트를 갱신할 수 있다.

zip 예제: 두 개의 배열 스트림을 순서대로 짝지어야 한다면, 먼저 of(1, 2, 3)과 of('a', 'b', 'c')를 준비하고, 그다음 zip(numbers$, letters$)로 구독한다. 결과는 [1, 'a'], [2, 'b'], [3, 'c'] 순서로 정확히 짝지어 방출된다. 인덱스 기준으로 데이터를 매칭해야 하는 상황에 적합하다.

forkJoin 예제: 여러 개의 API 요청이 모두 끝난 뒤 한 번에 결과를 처리하고 싶다면, 먼저 getUsers(), getPosts(), getComments()가 각각 HTTP 요청을 나타내는 옵저버블을 반환하게 하고, 그다음 forkJoin([getUsers(), getPosts(), getComments()])로 구독한다. 세 요청이 모두 완료되면 각각의 마지막 응답값이 배열로 한 번에 전달된다. 병렬 요청 후 결과를 합쳐서 화면을 그리는 초기 로딩 화면에 자주 쓰인다.

5. 실무 사용 사례

merge는 여러 UI 이벤트(버튼 클릭, 키보드 입력, 스크롤 등)를 하나의 스트림으로 모아서 공통 처리 로직(로그 기록, 자동 저장 트리거)을 적용할 때 사용한다.

concat은 순서가 중요한 API 호출 체인, 예를 들어 로그인 요청이 끝난 뒤 사용자 프로필을 조회해야 하는 경우, 혹은 여러 애니메이션을 순차적으로 재생해야 하는 경우에 사용한다.

combineLatest는 폼(form)의 여러 입력 필드 값을 조합해서 유효성 검사나 실시간 미리보기를 만들 때, 또는 여러 설정값이 바뀔 때마다 화면을 다시 계산해야 하는 대시보드에서 사용한다.

zip은 서로 다른 소스에서 온 데이터를 인덱스 기준으로 정확히 짝지어야 하는 경우, 예를 들어 여러 센서에서 동시에 수집된 데이터를 순서대로 매칭해야 하는 상황에 사용한다.

forkJoin은 화면 초기 진입 시 여러 개의 독립적인 API를 병렬로 호출하고, 모든 응답이 도착한 뒤에야 화면을 렌더링해야 하는 초기 로딩 로직에 가장 흔하게 사용된다.

6. 주의할 점

concat에 전달한 앞쪽 옵저버블이 완료되지 않는 무한 스트림이라면, 뒤에 있는 옵저버블은 절대로 실행되지 않는다. 완료 시점이 명확한 옵저버블에만 사용해야 한다.

combineLatest는 모든 소스가 최소 한 번은 값을 내야 첫 방출이 일어난다. 소스 중 하나가 값을 늦게 내거나 아예 값을 내지 않으면 결과 스트림도 아무것도 방출하지 않는다.

zip은 소스 간 발행 속도 차이가 크면 빠른 소스의 값이 내부 버퍼에 계속 쌓인다. 느린 소스가 있는 상황에서 장시간 사용하면 메모리 사용량이 늘어날 수 있다.

forkJoin은 소스 중 하나라도 완료되지 않으면 전체 결과가 영원히 나오지 않는다. 또한 소스 중 하나라도 에러를 내면 즉시 에러로 종료되고 다른 소스의 결과는 무시된다. HTTP 요청처럼 반드시 완료되는 옵저버블에만 사용하는 것이 안전하다.

merge는 값의 순서를 보장하지 않으므로, 순서가 중요한 로직에는 적합하지 않다.

7. 핵심 정리

merge는 순서 상관없이 값을 섞어서 흘려보내고, concat은 순서대로 하나씩 끝내고 넘어간다. combineLatest는 최신 값들의 조합을 계속 갱신해서 보여주고, zip은 인덱스 기준으로 정확히 짝지어 값을 방출한다. forkJoin은 모든 소스가 완료된 후 마지막 값들만 한 번에 모아준다. 다섯 연산자는 각각 "타이밍을 어떻게 다룰 것인가"에 대한 답이 다르므로, 요구사항이 순서 보장인지, 최신 상태 반영인지, 완료 후 일괄 처리인지를 먼저 파악한 뒤 선택해야 한다.