[TIL / 25.05.11] RxSwift.. 처음 공부해볼게요 2

https://subkyu-ios.tistory.com/45

[TIL / 25.05.08] RxSwift.. 처음 공부해볼게요 1

 

이어서 공부한 내용,,

---

RxSwift의 Operator(연산자)

Observable은 데이터가 흘러가는 강물이고, 연산자는 그 흐름을 가공하는 정수처리기라고 생각하면 됨

여기서 데이터는 언제 들어올지 모르는 데이터(비동기)고, 그렇다 보니 개수도 얼마나 될지 모르면서 이들을 우리는 매번 그대로 가져다 써야하는 아름다운 상황만 있는게 아니잖음?

 

이때 필요한 도구가 연산자(operators)임

 

연산자로는 예를 들어 map, filter, flatmap, zip, merge가 있는데 이 중 몇가지를 설명하면

 

 

map 

 

Observable에서 방출되는 값을 변형하고 싶을 때 사용. ex) 숫자를 텍스트로

Observable.of(1, 2, 3)
    .map { $0 * 2 }

 

이렇게 되면 이 Observable이라는 데이터 스트림은 1, 2, 3을 순서대로 방출하는데 거기에 2를 곱하는 장치를 넣어 2, 4, 6으로 데이터가 가공 될거임.

 

 

flatMap

 

방출된 값을 바탕으로 또 다른 Observable을 생성해야 할 때 사용. ex) userId -> 사용자 정보 요청, 좌표 -> 날씨 API 요청 등

이 연산자는 비동기 체이닝이 필요할 때, 그리고 값을 Observable로 바꿔서 이어서 작업해야 할 때 사용하면 좋음.

 

예를 들면 앱에서 api 요청을 연쇄적으로 해야 할 때가 있잖음? 어느 유저 블로그에 들어가면 그 유저 블로그 내에 다른 api 요청을 할 일이 생긴다. 그러면 해당 유저의 정보 중 id를 따와야 하기 때문에 여기서 첫 번째 비동기 작업. 거기서 따온 id로 요구되는 정보를 받기 위해 api 요청할 수 있도록 하는 두 번째 비동기 작업. 이런 경우에 씀.

(userId를 가져왔다고 쳐도 그걸 가지고 그냥 추가 정보를 가져온다는게 아니라 또 다른 네트워크 작업을 해야 하는 것이기에!)

 

userIDs
    .flatMap { fetchUser(id: $0) }

 

이 예시에선 아마 어느 그룹의 유저들의 id를 받아온게 userIDs고(첫 번째 작업) 각 요소마다 유저 정보를 받아오도록 하는 fetchUser 비동기 작업을 위해 flatMap을 써준 것.

 

zip

 

여러 Observable을 묶어 동시에 처리하고 싶을 때 사용. 각 Observable이 모두 하나씩 값을 내보낼 때만 다음 단계로 넘어감.

두 개 이상의 비동기 결과가 동시에 필요할 때 사용하면 좋음. ex) 이름 + 점수 -> (학생, 점수) 튜플

 

Observable.zip(names, scores)

 

names, scores에서 하나씩 각각 방출될 때마다 묶어서 출력됨. 짝을 맞추는 용도로 쓰임.

 

merge

 

여러 Observable을 동시에 처리하고, 그 결과를 한 스트림에 그대로 섞어보낼 때 사용.

각기 다른 Observable이 동시에 독립적으로 작동하는 상황에서, 결과를 하나로 합쳐서 보고 싶을 때 사용하면 좋음.

 

Observable.merge(subject1, subject2)

 

두 개의 Subject가 따로따로 값을 내보내지만, 결과는 하나의 흐름으로 이어서 받고자 할 때 사용됨. ex) 버튼 2개가 동일한 액션을 실행할 때

 

음.. 이건 살짝 납득이 잘 안되니 지선생에게 얼추 쓰임새를 물어보니

 

• 버튼 여러 개가 같은 작업을 트리거할 때
• 센서나 이벤트 소스를 동시에 모니터링할 때
• 비동기 로딩 두 개를 동시에 시작하고, 각각 결과를 받기 원할 때

쓴다고 하네요~

 

---

 

그럼 Observable을 만들고 연산자로 데이터 가공하는 법까지 배워봤으니 이를 어느 스레드에서 할지도 배워봅시다

 

Scheduler

Scheduler란, Observable 스트림에서 스레드를 지정할 수 있도록 돕는 도구.

예를 들어, Observable 구독을 통해서 UI 작업을 처리하게 되었다면, 그 작업이 메인 스레드에서 동작하길 명시하는 편이 좋음.

 

• MainSchedular
  • 메인 스레드에서 작업을 실행하는 스케줄러
  • 주로 UI 업데이트와 관련된 작업을 실행할 때 사용
  • ex) MainScheduler.instance
• ConcurrentDispatchQueueSchedular
  • 비동기 작업을 위한 DispatchQueue 기반의 스케줄러 (아 디스패치큐 다시 복습해봐야겠다)
  • 여러 스레드에서 동시 작업을 처리할 수 있음
  • ex) ConcurrentDispatchQueueScheduler(qos: .background)
    • qos는 우선순위 레벨인데 (이건 좀 이것저것 써봐야 쓰임새를 잘 알 수 있을듯..?)
      • userInteractive: 유저가 누르면 즉각 반응 (main queue)
      • userInitiated: 유저가 실행시킨 작업들을 즉각적이지는 않지만, async하도록 처리
      • default
      • utility: I/O, n/w API 호출
      • background: 유저가 인지하지 못할 정도의 뒷단에서 수행하는 작업
• SerialDispatchQueueScheduler
  • 특정 DispatchQueue에서 직렬 작업을 실행하는 스케줄러
  • 동시성이 필요 없는 작업을 특정 스레드에서 순서대로 처리할 때 유용
  • Serial이라는 키워드를 생각해보면 좋은데, 이 부분은 나도 헷갈리니 다시 볼 때 공부할 수 있게 메모 좀..
    • https://babbab2.tistory.com/64

 

subscribeOn & observeOn

 

subscribeOn과 observeOn은 옵저버블이 처리되어야 할 스레드를 지정하는 역할을 함.

• subscribeOn : 스트림이 시작되는 스레드를 지정
• observeOn : 다운스트림의 스레드를 지정함 - 그 이후 흐름(연산자 체인)의 스레드를 지정

 

 

import Foundation
import RxSwift
import RxCocoa

struct MyError: Error {}

let disposeBag = DisposeBag()

Observable<Int>.create { observer in
    print("🔥 create: \(Thread.current)")
    observer.onNext(1)
    return Disposables.create()
}
.map {
    print("🧮 map: \(Thread.current)")
    $0 * 2
}
//.subscribe(on: ConcurrentDispatchQueueScheduler(qos: .background))
.subscribe(on: MainScheduler.instance)
.subscribe(onNext: {
    print("✅ onNext: \(Thread.current)")
})

let read = readLine()

 

subscribe on을 통해 메인스레드에서 실행되도록 지정해줬고 observe on이 있지 않는 이상 모두 그렇게 작동함을 알 수 있음

 

다만 subscribe on이 모든 경우에 적용되지만은 않는데, Hot observable같이 이미 스트림이 시작되어있는 경우엔 그 뒤에 제어해줄 수가 없음

 

import Foundation
import RxSwift

let disposeBag = DisposeBag()

let observable = Observable<Int>
    .interval(.seconds(1), scheduler: MainScheduler.instance)
    .do(onNext: { value in
        print("🔥 Emitted: \(value), Thread: \(Thread.current)")
    })

observable
    .subscribe(on: ConcurrentDispatchQueueScheduler(qos: .background)) // ✅ 무시됨
    .observe(on: ConcurrentDispatchQueueScheduler(qos: .background))   // ✅ 적용됨
    .map { value in
        print("🧮 map: \(value), Thread: \(Thread.current)")
        return value * 2
    }
    .observe(on: MainScheduler.instance) // 다시 메인 스레드로 전환
    .subscribe(onNext: {
        print("✅ onNext: \($0), Thread: \(Thread.current)")
    })
    .disposed(by: disposeBag)

RunLoop.main.run()

 

interval의 경우엔 바로 방출을 시작하기에 스트림이 시작되어있어서 스레드가 이미 정해져있고 그 뒤에 subscribe on을 한다고 한들 적용되지 않음. 즉, subscribe on은 스트림의 시작 스레드를 지정한다고 해줬잖아요? 근데 이미 물은 흐르기 시작해서 그 뒤에 설정해줄 수 없다는 것

 

Hot observable의 예시로 button.rx.tap이 있는데 이 경우에도 UI 이벤트 Observable이라 메인 스레드에서 발생해, RxCocoa내부에서 이미 MainScheduler에 바인딩 되어있어서 subscribe on하는 의미가 없음.

 

이렇게 봤을 때 subscribe on은 주로 cold observable에서 유의미하게 작동하는구나 생각할 수 있음. 근데 이게 맞으면 블로그들에서 다뤘을텐데 관련 글이 안보이는걸 보니 아닌가..? 싶은데 이건 한번 여쭤보겠음

 

---

 

Subject, Relay

 

Subject

 

subject는 Observable과 Observer의 역할을 모두 수행할 수 있음

대표적으론 BehaviorSubject, PublishSubject가 있는데

 

BehaviorSubject : 초깃값을 가지며, 구독을 시작하면 가장 최근에 방출되었던 값을 받으며 구독을 시작함

let disposeBag = DisposeBag()

let subject = BehaviorSubject(value: 0)

// Observer 의 역할도 수행한다.
// Observer 의 역할이란, 값을 "받아보는" 입장임.
// subject 가 10 이라는 값을 받아보고 있음.
subject.onNext(10)

// Observable 역할이기 때문에 당연히 구독 가능.
subject.subscribe(onNext: { data in
    print("onNext: \(data)")
}).disposed(by: disposeBag)

subject.onNext(20)

subject.onNext(30)

구독 시에 초깃값으로 10 받고 그 뒤에 20, 30 방출해 onNext 처리해줌

 

PublishSubject : 초깃값을 가지지 않으며, 구독을 시작했어도 가장 최근에 방출되었던 값을 받지 않음. 구독 이후로 흐른 값만을 받아봄

let disposeBag = DisposeBag()

let subject = PublishSubject<Int>()

subject.onNext(10)

subject.subscribe(onNext: { data in
    print("onNext: \(data)")
}).disposed(by: disposeBag)

subject.onNext(20)

subject.onNext(30)

 

Relay

 

Relay도 Subject와 비슷하게 Observable의 역할과 Observer의 역할을 겸함다만 에러나 완료 이벤트를 방출하지 않도록 설계된 RxCocoa의 객체고, 에러나 완료가 되지 않기 때문에 주로 UI 이벤트 처리에서 사용되며(UI 그리는 작업은 멈추면 안되기 때문), 애플리케이션 상태 관리에 유용함.

 

BehaviorRelay : 초깃값을 가지며, 구독을 시작하면 가장 최근에 방출되었던 값을 받으며 구독을 시작함

import RxCocoa // 반드시 import 해줄 것.

let disposeBag = DisposeBag()

let relay = BehaviorRelay(value: 0)

// Observer 의 역할도 수행한다.
// Observer 의 역할이란, 값을 "받아보는" 입장임.
// subject 가 10 이라는 값을 받아보고 있음.
relay.accept(10)

// Observable 역할이기 때문에 당연히 구독 가능.
relay.subscribe(onNext: { data in
    print("onNext: \(data)")
}).disposed(by: disposeBag)

relay.accept(20)

relay.accept(30)

 

subject같은 다른 observable과 얘는 좀 다른데 accept라는 키워드를 사용해서 이벤트를 방출함.

 

 

PublishRelay : PublishSubject와 논리가 같음. 초깃값 X, 구독 시작했어도 최근 방출된 값 받지않고 이후로 흐른 값만을 받음.

import RxCocoa // 반드시 import 해줄 것.

let disposeBag = DisposeBag()

let relay = PublishRelay<Int>()

// Observer 의 역할도 수행한다.
// Observer 의 역할이란, 값을 "받아보는" 입장임.
// subject 가 10 이라는 값을 받아보고 있음.
relay.accept(10)

// Observable 역할이기 때문에 당연히 구독 가능.
relay.subscribe(onNext: { data in
    print("onNext: \(data)")
}).disposed(by: disposeBag)

relay.accept(20)

relay.accept(30)

 

그럼 이 Relay를 쓰기 위해선 RxCocoa를 import해야 한다고 했는데 RxCocoa가 뭔데요?

 

 

RxCocoa

RxSwift는 Swift에 대한 Rx 프로그래밍을 지원한다면, RxCocoa는 iOS, macOS에 대해서 Rx 프로그래밍을 지원하는 도구임.

(RxSwift엔 UIKit에 대한 정보가 없음)

 

import UIKit
import SnapKit
import RxSwift
import RxCocoa

class ViewController: UIViewController {

    let disposeBag = DisposeBag()
    
    let button: UIButton = {
        let button = UIButton()
        button.backgroundColor = .blue
        button.setTitle("버튼", for: .normal)
        return button
    }()
    
    override func viewDidLoad() {
        super.viewDidLoad()
        initUI()
        bind()
    }

    func initUI() {
        view.backgroundColor = .white
        [button].forEach { view.addSubview($0) }
        
        button.snp.makeConstraints {
            $0.width.equalTo(120)
            $0.height.equalTo(80)
            $0.center.equalToSuperview()
        }
    }
    
    func bind() {
        button.rx.tap
            .subscribe(onNext: { _ in
                print("버튼이 클릭되었음.")
            }).disposed(by: disposeBag)
    }
}

 

button.rx.tap을 하나의 스트림으로 간주해 구독하는 코드를 작성 가능.

 

이게 좋은게 쓰로틀링이라고 설정한 시간 간격 내로 한번씩만 이벤트를 방출하도록 할 수가 있는데 원랜 따로 구현을 해줘야하고 연동까지 해줘야하는데 여기선 .throttle(.seconds(1), scheduler: MainScheduler.instance)같은 코드 하나로 해결이 됨.

 

이외에도 이점이 많지만 일단 개념공부는 여기까지 하고 과제 풀면서 적용한 바를 따로 정리해보겠음~

 

---

 

Hot, Cold 옵저버블에 대해 스트림의 시작 시점과 subscribe on의 유의미성 그리고 이 메서드를 미리 호출해야된다는 gpt의 말이 있었는데 계속 파고파다가 이틀동안 굴레에 갇혀있었음,, 제대로 된 결론이 없어서 많이 아쉬운데 경험해보면서 하나씩 뽑아보자,, 화이팅