[RxSwift] RxSwift를 이루는 Observable과 Observer

2025.02.25 - [/RxSwift] - [RxSwift] RxSwift 란 무엇일까 ?

 

 

안녕하세요 아렉스입니다 :>
오늘은 RxSwift를 이루는 주요 개념인 Observable과 Observer에 대해서 알아보겠습니다.

 

Observable

---

Observable은 데이터를 방출(emit)하는 스트림(Stream)입니다.

Stream 의 사전적 의미는 시냇물, 흐름, 줄기 라는 의미를 갖습니다. 

 

Observable 은 데이터의 흐름이며, 시간이 지남에 따라 이산적인 값을 방출하는 방식으로 작동한다는 것입니다.

즉, Observable에는 시간적인 개념이 포함이 됩니다.

네트워크 요청, 사용자 입력, 타이머 등의 데이터를 이벤트 형태로 방출하고, 이를 Observer(구독자) 가 받아 처리할 수 있도록 합니다.

Observable의 특징

• 이벤트를 연속적으로 방출할 수 있습니다.
• 비동기적으로 동작하며, 구독(Subscription)한 옵저버에게 데이터를 전달합니다.
• 여러 개의 Observer가 동일한 Observable을 구독할 수 있습니다.
• 3가지 이벤트 타입을 방출할 수 있습니다.
  • onNext(value): 새로운 데이터를 방출할 때마다 호출
  • onError(error): 오류 발생 시 오류를 방출하고 스트림 종료(dispose 호출); onNext나 onCompleted 는 더 이상 호출 되지않는다.
  • onCompleted(): 정상적으로 작업(onNext 호출)이 끝나면 스트림 종료(dispose 호출)

onError와 onCompleted 는 자신의 이벤트가 발생한 뒤 스트림을 종료하는 것을 기억하고 넘어갑시다

 

Observer

---

Observer는 Observable이 방출하는 데이터를 구독하고 처리하는 역할을 합니다.

하지만 Observer는 구현체가 아니라 개념적인 요소입니다.

실제 옵저버 역할은 subscribe 와 같은 Observable 과 Observer를 연결해주는 메서드 내부 클로저를 통해 수행됩니다.

Observer의 특징

• Observable의 이벤트를 수신하고 처리합니다.
• onNext, onError, onCompleted 이벤트를 처리할 수 있습니다.
• 실제 옵저버 역할은 subscribe 내부에서 정의됩니다.

 

정리

---

• Observable은 데이터를 방출하는 역할을 합니다.
• Observer는 Observable이 방출하는 데이터를 구독하고 처리하는 역할을 합니다.
• subscribe를 호출하면 Observer 역할을 하는 클로저가 이벤트를 처리하게 됩니다.

전체 흐름

1. Observable이 이벤트를 방출 (onNext, onError, onCompleted)
2. Observer가 subscribe를 통해 이벤트를 구독
3. 이벤트 발생 시 Observer의 클로저가 실행

 

예제 코드를 통해 이해해보겠습니다.

print() 문을 활용하면 원하는 값을 볼 수 있지만, RxSwift에는 debug() 연산자 또한 지원합니다.

Observable의 스트림과 이벤트 부분을 더 편하게 디버깅할 때 용이합니다.

// ✅ 1. Observable 생성
let create = Observable.create { observer in
    // ✅ 2. onNext(_:) 이벤트 발생
    observer.onNext("red")
    observer.onNext("blue")
    observer.onNext("yellow")
    return Disposables.create()
}

create
    .debug() // ✅ 3. debug() 연산자 사용
    .subscribe(onNext: { data in
        // ✅ 4. 1,2번 에서 발생한 이벤트를 수신하는 Observer
        print("onNext: \(data)")
    }, onError: { error in
        print("onError: \(error)")
    }, onCompleted: { 
        print("onCompleted: ✅")
    })
    .disposed(by: diposeBag)
    
// ✅ 5. 실제 출력 확인
// 2025-02-26 16:47:32.401: ViewController.swift:42 (viewDidLoad()) -> subscribed
// 2025-02-26 16:47:32.401: ViewController.swift:42 (viewDidLoad()) -> Event next(red)
// onNext: red
// 2025-02-26 16:47:32.401: ViewController.swift:42 (viewDidLoad()) -> Event next(blue)
// onNext: blue
// 2025-02-26 16:47:32.401: ViewController.swift:42 (viewDidLoad()) -> Event next(yellow)
// onNext: yellow

 

 

onError와 onCompleted 는 자신의 이벤트가 발생한 뒤 스트림을 종료하는 것을 기억하고 계신가요 ?

 

현재 Observable이 방출한 값에 따라서 onNext 이벤트가 전달 되는 모습을 볼 수있습니다.

하지만 onCompleted 혹은 onError 가 호출이 되지않았기에 스트림은 종료가 되지않은 것을 볼 수 있습니다.

RxSwift 내부적으로 연산자에 따라 방출할 스트림이 없을 때 자동으로 onCompleted를 호출하기도합니다

하지만 Observable을 생성해주는 create 연산자는 onCompleted()를 별도로 호출해야합니다.

 

onCompleted() 호출이 추가 된 완성코드

// ✅ 1. Observable 생성
let create = Observable.create { observer in
    // ✅ 2. onNext(_:) 이벤트 발생
    observer.onNext("red")
    observer.onNext("blue")
    observer.onNext("yellow")
    observer.onCompleted() // onCompleted 호출
    return Disposables.create()
}

create
    .debug() // ✅ 3. debug() 연산자 사용
    .subscribe(onNext: { data in
        // ✅ 4. 1,2번 에서 발생한 이벤트를 수신하는 Observer
        print("onNext: \(data)")
    }, onError: { error in
        print("onError: \(error)")
    }, onCompleted: { 
        print("onCompleted: ✅")
    })
    .disposed(by: diposeBag)
    
// ✅ 5. 실제 출력 확인
// 2025-02-26 16:47:32.401: ViewController.swift:42 (viewDidLoad()) -> subscribed
// 2025-02-26 16:47:32.401: ViewController.swift:42 (viewDidLoad()) -> Event next(red)
// onNext: red
// 2025-02-26 16:47:32.401: ViewController.swift:42 (viewDidLoad()) -> Event next(blue)
// onNext: blue
// 2025-02-26 16:47:32.401: ViewController.swift:42 (viewDidLoad()) -> Event next(yellow)
// onNext: yellow
// 2025-02-26 16:47:32.401: ViewController.swift:43 (viewDidLoad()) -> Event completed
// onCompleted: ✅
// 2025-02-26 16:47:32.401: ViewController.swift:43 (viewDidLoad()) -> isDisposed

 

 

시퀀스 종료

---

시퀀스가 onCompleted, onError, 이벤트가 보내면 모든 스트림(내부 리소스)가 종료가 됩니다.

시퀀스 요소(element)의 생성을 중단하고 리소스를 즉시 해제하려면 반환 된 구독에서 폐기(dispose)를 요청하세요.

 

스트림이 종료가 되지않는다는 것은 향후에 메모리 누수나 리소스 해제가 제대로 이루어지지 않을 수 있기 때문에 중요한 부분입니다.

구독을 명시적으로 종료하지 않으면, 스트림을 계속 활성 상태로 남아있고, 그에 따라 메모리 리소스도 해제되지않습니다.

이로 인해 메모리 누수가 발생할 수 있습니다.

 

따라서 스트림을 정상적으로 종료 혹은 dispose를 호출하여 구독을 종료하여 시퀀스가 더 이상 자원을 사용하지않도록 해야합니다.

 

 

https://reactivex.io/documentation/ko/observable.html

ReactiveX - Observable

https://github.com/ReactiveX/RxSwift/blob/main/Documentation/GettingStarted.md#sharing-subscription-and-share-operator