RxSwift для Swift: полное руководство с примерами

Что такое RxSwift и зачем он нужен?

RxSwift — это официальная реализация ReactiveX (Reactive Extensions) для языка Swift. ReactiveX — это библиотека для асинхронного программирования и манипуляции потоками данных с использованием наблюдаемых последовательностей (Observable) и операторов (Operators). RxSwift позволяет писать чистый, декларативный и легко читаемый код для обработки событий, асинхронных операций (сетевые запросы, работа с UI, таймеры) и управления состоянием приложения.

Зачем нужен RxSwift?

• Упрощение асинхронного кода: Вместо вложенных замыканий (callback hell) и делегатов, RxSwift предлагает единый интерфейс для работы с любыми асинхронными источниками данных.
• Декларативность: Вы описываете что должно произойти с данными, а не как это реализовать. Код становится более предсказуемым.
• Управление потоками (Threading): Легко переключаться между фоновыми и главным потоками с помощью операторов observeOn и subscribeOn.
• Композиция: Вы можете комбинировать, фильтровать, трансформировать и объединять несколько потоков данных в один.
• Управление памятью: Встроенные механизмы (DisposeBag) автоматически отписывают подписки при уничтожении объектов, предотвращая утечки памяти.

RxSwift — это не просто библиотека, это парадигма программирования, которая особенно полезна в сложных iOS-приложениях с множеством взаимодействий и состояний.

Установка RxSwift

RxSwift устанавливается через CocoaPods или Swift Package Manager (SPM). Установка через pip или npm не применима, так как это нативная библиотека для экосистемы Apple.

Установка через CocoaPods

Добавьте в ваш Podfile:

platform :ios, '13.0'
use_frameworks!

target 'YourAppTarget' do
  pod 'RxSwift', '~> 6.0'
  pod 'RxCocoa', '~> 6.0'  # Для интеграции с UIKit/AppKit
end

Затем выполните команду в терминале:

pod install

Установка через Swift Package Manager (SPM)

В Xcode: File → Add Packages... и вставьте URL репозитория:

https://github.com/ReactiveX/RxSwift.git

Выберите нужную версию (например, 6.5.0) и добавьте пакет RxSwift и RxCocoa в ваш таргет.

Быстрый старт с примером кода

Создайте простой Observable, который эмитирует числа от 1 до 5, и подпишитесь на него:

import RxSwift

let disposeBag = DisposeBag()

// 1. Создаем Observable
let observable = Observable<Int>.create { observer in
    observer.onNext(1)
    observer.onNext(2)
    observer.onNext(3)
    observer.onNext(4)
    observer.onNext(5)
    observer.onCompleted()
    return Disposables.create()
}

// 2. Подписываемся
observable
    .subscribe(
        onNext: { value in
            print("Получено значение: \(value)")
        },
        onError: { error in
            print("Ошибка: \(error.localizedDescription)")
        },
        onCompleted: {
            print("Последовательность завершена")
        },
        onDisposed: {
            print("Подписка уничтожена")
        }
    )
    .disposed(by: disposeBag)

// Вывод:
// Получено значение: 1
// Получено значение: 2
// Получено значение: 3
// Получено значение: 4
// Получено значение: 5
// Последовательность завершена
// Подписка уничтожена

Объяснение:

• Observable.create — создает кастомную последовательность.
• observer.onNext — эмитирует новое значение.
• observer.onCompleted — сигнализирует о завершении последовательности.
• subscribe — подписывается на события.
• disposed(by: disposeBag) — добавляет подписку в мешок для автоматической отписки.

Основные методы и операторы RxSwift

RxSwift предлагает десятки операторов. Ниже приведены 10 самых важных и часто используемых.

1. map

Что делает: Трансформирует каждое значение из Observable в новое значение.

Параметры: Замыкание, которое принимает элемент типа T и возвращает элемент типа R.

Пример:

let numbers = Observable.of(1, 2, 3, 4, 5)
numbers
    .map { $0 * 10 }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
// Вывод: 10, 20, 30, 40, 50

2. filter

Что делает: Пропускает только те значения, которые удовлетворяют условию.

Параметры: Замыкание-предикат, возвращающее Bool.

Пример:

let numbers = Observable.of(1, 2, 3, 4, 5, 6)
numbers
    .filter { $0 % 2 == 0 }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
// Вывод: 2, 4, 6

3. flatMap

Что делает: Проецирует каждое значение в новый Observable и "сплющивает" все полученные Observable в один поток. Используется для работы с асинхронными операциями (например, сетевые запросы).

Параметры: Замыкание, которое принимает элемент и возвращает Observable<R>.

Пример:

let userIds = Observable.of(1, 2, 3)
userIds
    .flatMap { userId in
        return fetchUserData(for: userId) // Возвращает Observable<User>
    }
    .subscribe(onNext: { user in
        print("User: \(user.name)")
    })
    .disposed(by: disposeBag)

4. merge

Что делает: Объединяет несколько Observable в один, сохраняя порядок эмиссии по времени.

Параметры: Массив или вариативный список Observable одного типа.

Пример:

let first = Observable.of("A", "B", "C")
let second = Observable.of("1", "2", "3")
Observable.merge(first, second)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
// Вывод: A, B, C, 1, 2, 3 (порядок может отличаться в зависимости от времени эмиссии)

5. combineLatest

Что делает: Объединяет последние значения из нескольких Observable и эмитирует новое значение при изменении любого из них. Требует, чтобы каждый Observable эмитировал хотя бы одно значение.

Параметры: Два или более Observable и замыкание-комбинатор.

Пример:

let username = Observable<String>.of("user1", "user2")
let password = Observable<String>.of("pass123", "pass456")
Observable.combineLatest(username, password) { (user, pass) in
    return "\(user):\(pass)"
}
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
// Вывод: user2:pass456 (последние значения обоих)

6. zip

Что делает: Объединяет значения из нескольких Observable попарно (по индексу). Эмитирует новое значение, когда все Observable предоставили следующий элемент.

Параметры: Два или более Observable и замыкание-комбинатор.

Пример:

let names = Observable.of("Alice", "Bob", "Charlie")
let scores = Observable.of(95, 87, 92)
Observable.zip(names, scores) { (name, score) in
    return "\(name): \(score)"
}
.subscribe(onNext: { print($0) })
.disposed(by: disposeBag)
// Вывод: Alice: 95, Bob: 87, Charlie: 92

7. debounce (throttle)

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

Параметры: dueTime (RxTimeInterval) — время ожидания.

Пример:

let searchText = PublishSubject<String>()
searchText
    .debounce(.milliseconds(300), scheduler: MainScheduler.instance)
    .subscribe(onNext: { text in
        print("Поиск: \(text)")
    })
    .disposed(by: disposeBag)

searchText.onNext("S")
searchText.onNext("Sw")
searchText.onNext("Swi")
// Пауза 300 мс
// Вывод: Поиск: Swi

8. distinctUntilChanged

Что делает: Пропускает только те значения, которые отличаются от предыдущего. Избегает дублирования одинаковых последовательных событий.

Параметры: Необязательное замыкание-компаратор.

Пример:

let numbers = Observable.of(1, 1, 2, 2, 3, 1, 1)
numbers
    .distinctUntilChanged()
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
// Вывод: 1, 2, 3, 1

9. catchError (catch)

Что делает: Перехватывает ошибку в Observable и позволяет продолжить последовательность, вернув запасной Observable или значение.

Параметры: Замыкание, которое принимает ошибку и возвращает новый Observable.

Пример:

let failingObservable = Observable<Int>.create { observer in
    observer.onNext(1)
    observer.onError(NSError(domain: "Test", code: -1))
    observer.onNext(2) // Не будет вызван
    return Disposables.create()
}
failingObservable
    .catchError { error in
        print("Ошибка: \(error). Возвращаем запасное значение")
        return Observable.just(0)
    }
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)
// Вывод: 1, Ошибка: ..., 0

10. observeOn и subscribeOn

Что делает: Управляют тем, на каком потоке (Scheduler) выполняются операции.

• subscribeOn — задает поток, на котором создается и эмитирует Observable.
• observeOn — задает поток, на котором вызываются обработчики подписки (onNext, onError и т.д.).

Параметры: Экземпляр SchedulerType (например, MainScheduler.instance, ConcurrentDispatchQueueScheduler).

Пример:

let backgroundScheduler = ConcurrentDispatchQueueScheduler(qos: .background)
Observable.just(1)
    .subscribeOn(backgroundScheduler) // Работаем в фоне
    .observeOn(MainScheduler.instance) // Результат получаем на главном потоке
    .subscribe(onNext: { value in
        print("На главном потоке: