RxJava

Определяет поток, на котором будет выполнено создание Observable и запуск его операций. Влияет на операторы выше. Если указан только subscribeOn - все операторы будут выполняться в указанном потоке.

Observable.just("item")
    .subscribeOn(Schedulers.io()) // Выполнение Observable на IO-потоке
    .observeOn(AndroidSchedulers.mainThread()) // Обработка результатов на основном потоке
    .subscribe(
        item -> System.out.println("Received: " + item),
        Throwable::printStackTrace
    );

Определяет поток, на котором будет производиться обработка элементов и уведомление об их изменении. Влияет на операторы ниже. Если указан только observeOn - все операторы будут выполняться в текущем потоке, а операторы, расположенные ниже observeOn будут переключаться на поток, указанный в observeOn.

Observable.just("item")
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.computation()) // Обработка результатов на вычислительном потоке
    .map(item -> item.toUpperCase())
    .observeOn(AndroidSchedulers.mainThread()) // Обработка окончательных результатов на основном потоке
    .subscribe(
        item -> System.out.println("Processed: " + item),
        Throwable::printStackTrace
    );

Действие с каждым элементом. Выполняет указанное действие для каждого элемента, который проходит через Observable, перед тем как он будет передан дальше.

observable.doOnNext { item -> println("Item: $item") }

Действие при завершении потока. Выполняет указанное действие, когда Observable завершает эмитирование всех элементов. Не вызывается, если поток был прерван с ошибкой.

observable.doOnComplete { println("Stream completed") }

Действие при ошибке. Выполняет указанное действие, когда возникает ошибка в процессе работы Observable.

observable.doOnError { error -> println("Error occurred: $error") }

Действие при подписке. Выполняет указанное действие, когда подписка на Observable начинается.

observable.doOnSubscribe { disposable -> println("Subscribed: $disposable") }

Действие после завершения работы Observable. Выполняет указанное действие после завершения работы Observable, независимо от того, завершился ли поток успешно или с ошибкой.

observable.doAfterTerminate { println("Stream terminated") }

Действие при отписке. Выполняет указанное действие, когда Disposable объект используется для отписки от Observable.

observable.doOnDispose { println("Disposed") }

Создает Observable, который излучает один элемент.

Observable.just("Hello, World!")
    .subscribe(System.out::println);

Создает Observable из коллекции.

List<String> list = Arrays.asList("One", "Two", "Three");
Observable.fromIterable(list)
    .subscribe(System.out::println);

Применяет функцию к каждому элементу.

Observable.just(1, 2, 3)
    .map(i -> i * 2)
    .subscribe(System.out::println);  // Output: 2, 4, 6

Преобразует каждый элемент в Observable и объединяет результаты. Порядок генерированных элементов не гарантируется.

Observable.just(1, 2, 3)
    .flatMap(i -> Observable.just(i, i * 2))
    .subscribe(System.out::println);  // Output: 1, 2, 2, 4, 3, 6

Преобразует каждый элемент в Observable и объединяет результаты, сохраняя порядок.

Observable.just(1, 2, 3)
    .concatMap(i -> Observable.just(i, i * 2))
    .subscribe(System.out::println);  // Output: 1, 2, 2, 4, 3, 6

Переключается на новый Observable каждый раз, когда приходит новый элемент, отменяя предыдущие Observable.

Observable<String> searchRequests = Observable.just("query1", "query2", "query3");

searchRequests
    .switchMap(query -> performSearch(query))
    .subscribe(
        result -> System.out.println("Result: " + result),
        Throwable::printStackTrace,
        () -> System.out.println("Completed")
    );

Observable<String> performSearch(String query) {
    return Observable.just("Results for " + query)
                     .delay(1, TimeUnit.SECONDS);
}

Пропускает элементы, которые не удовлетворяют условию.

Observable.just(1, 2, 3, 4, 5)
    .filter(i -> i % 2 == 0)
    .subscribe(System.out::println);  // Output: 2, 4

Пропускает первые N элементов в Observable и затем передает оставшиеся элементы.

Observable<Integer> source = Observable.range(1, 5);
source.skip(2) // Пропускает первые 2 элемента
    .subscribe(System.out::println); // Output: 3, 4, 5

Этот оператор пропускает последние N элементов в Observable и передает все остальные.

Observable<Integer> source = Observable.range(1, 5);
source.skipLast(2) // Пропускает последние 2 элемента
    .subscribe(System.out::println); // Output: 1, 2, 3

Берет только первые count элементов.

Observable.just(1, 2, 3, 4, 5)
    .take(3)
    .subscribe(System.out::println);  // Output: 1, 2, 3

Берет последние count элементов.

Observable.just(1, 2, 3, 4, 5)
    .takeLast(3)
    .subscribe(System.out::println);  // Output: 3, 4, 5

Объединяет несколько Observable в один.

Observable.merge(
    Observable.just(1, 2),
    Observable.just(3, 4)
).subscribe(System.out::println);  // Output: 1, 2, 3, 4

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

Observable.concat(
    Observable.just(1, 2),
    Observable.just(3, 4)
).subscribe(System.out::println);  // Output: 1, 2, 3, 4

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

Observable.zip(
    Observable.just(1, 2),
    Observable.just("A", "B"),
    (i, s) -> i + s
).subscribe(System.out::println);  // Output: 1A, 2B

Комбинирует последние значения из нескольких Observable при каждом обновлении любого из них.

Observable.combineLatest(
    Observable.just(1, 2).startWith(0),
    Observable.just("A", "B"),
    (i, s) -> i + s
).subscribe(System.out::println);  // Output: 0A, 1A, 1B, 2B

Объединяет элементы из двух Observable на основе определенного условия.

Observable<String> source1 = Observable.just("A", "B", "C");
Observable<Integer> source2 = Observable.just(1, 2, 3);

source1.join(
    source2,
    item -> Observable.timer(1, TimeUnit.SECONDS), // Соединение через Observable
    item -> Observable.timer(1, TimeUnit.SECONDS), // Соединение через Observable
    (s, i) -> s + i // Комбинирование элементов
).subscribe(System.out::println);

Возвращает новый Observable при возникновении ошибки.

Observable.just(1, 2, 3)
    .map(i -> {
        if (i == 2) throw new RuntimeException("Error!");
        return i;
    })
    .onErrorResumeNext(Observable.just(10, 20))
    .subscribe(System.out::println);  // Output: 1, 10, 20

Возвращает заданный элемент при возникновении ошибки.

Observable.just(1, 2, 3)
    .map(i -> {
        if (i == 2) throw new RuntimeException("Error!");
        return i;
    })
    .onErrorReturnItem(-1)
    .subscribe(System.out::println);  // Output: 1, -1

Повторяет попытку при возникновении ошибки.

Observable.just(1, 2, 3)
    .map(i -> {
        if (i == 2) throw new RuntimeException("Error!");
        return i;
    })
    .retry(1)
    .subscribe(System.out::println);  // Output: 1, 2, 3

Позволяет управлять повторными попытками при возникновении ошибки.

Observable.just(1, 2, 3)
    .map(i -> {
        if (i == 2) throw new RuntimeException("Error!");
        return i;
    })
    .retryWhen(errors -> errors.take(1).flatMap(error -> Observable.just(4)))
    .subscribe(System.out::println);  // Output: 1, 2, 4, 3

Задерживает эмиссию элементов.

Observable.just(1, 2, 3)
    .delay(1, TimeUnit.SECONDS)
    .subscribe(System.out::println);  // Output: 1, 2, 3 (через 1 секунду)

Эмитирует элементы, которые не были заменены другим элементом в течение заданного времени.

Observable.create(emitter -> {
    emitter.onNext("A");
    Thread.sleep(100);
    emitter.onNext("B");
    Thread.sleep(100);
    emitter.onNext("C");
    emitter.onComplete();
}).debounce(150, TimeUnit.MILLISECONDS)
  .subscribe(System.out::println);  // Output: C

Применяет функцию накопления к каждому элементу и предыдущему результату.

Observable.just(1, 2, 3)
    .scan(0, (total, next) -> total + next)
    .subscribe(System.out::println);  // Output: 0, 1, 3, 6

Удаляет дублирующиеся элементы.

Observable.just(1, 1, 2, 2, 3, 3)
    .distinct()
    .subscribe(System.out::println);  // Output: 1, 2, 3

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

val observable = Observable.just(1, 2, 3)
val sharedObservable = observable.share()

sharedObservable.subscribe { println("Subscriber 1: $it") }
sharedObservable.subscribe { println("Subscriber 2: $it") }

Cоздает Observable, который через указанное время эмитит один элемент и завершает выполнение.

Observable.timer(5, TimeUnit.SECONDS)
    .subscribe { println("Completed after 5 seconds") }

Используется для указания максимального времени ожидания между элементами потока. Если за это время не поступает новый элемент, поток выбрасывает ошибку TimeoutException.

Observable
    .just(1, 2, 3)
    .delay(2, TimeUnit.SECONDS) // задержка
    .timeout(1, TimeUnit.SECONDS) // если больше 1 секунды ожидания, выбросит ошибку
    .subscribe(
        { item -> println("Received: $item") },
        { error -> println("Error: $error") }
    )

Источник данных, который испускает события или элементы данных и позволяет наблюдателям (Observers) подписываться на них. Это основной компонент для работы с реактивным программированием.

Observable.just("Hello", "RxJava")
    .map { it.uppercase() }
    .filter { it.startsWith("H") }
    .subscribe(
        { println("Received: $it") },
        { println("Error: ${it.message}") },
        { println("Completed!") }
    )

Тип Observable, который испускает только одно значение или ошибку. Используется, когда нужно вернуть одно значение, например, результат сетевого запроса.

Single.just("Hello, RxJava")
    .map { it.uppercase() }
    .subscribe(
        { println("Success: $it") },
        { println("Error: ${it.message}") }
    )

Тип Observable, который не испускает данные, а лишь сигнализирует о завершении (onComplete) или ошибке (onError). Используется для выполнения задач без возвращаемого результата, например, операций записи или удаления.

Completable.fromAction {
    println("Executing some action...")
}
    .subscribe(
        { println("Action completed!") },
        { error -> println("Error: ${error.message}") }
    )

Источник, который может либо вернуть одно значение, либо завершиться без результата, либо сообщить об ошибке. Используется для операций, где результат может быть необязательным, например, получение данных из кэша.

Maybe.just("Hello, Maybe")
    .filter { it.length > 5 }
    .subscribe(
        { println("Success: $it") },
        { println("Error: ${it.message}") },
        { println("No data emitted") }
    )

Предоставляет пул потоков для задач ввода-вывода, таких как сетевые запросы, чтение файлов или работу с базой данных. Пул автоматически масштабируется под нагрузку.

fun main() {
    Observable.just("File 1", "File 2")
        .subscribeOn(Schedulers.io())
        .subscribe { println("Processing: $it on ${Thread.currentThread().name}") }
}

Предоставляет пул потоков, оптимизированный для вычислительных задач, таких как обработка данных или сложные расчёты. Количество потоков равно числу процессоров устройства.

fun main() {
    Observable.range(1, 5)
        .subscribeOn(Schedulers.computation())
        .map { it * it } // Вычисление квадрата
        .subscribe { println("Result: $it on ${Thread.currentThread().name}") }
}

Выполняет все задачи на одном общем потоке, предназначенном для последовательного выполнения. Это полезно для операций, где важна строгая последовательность выполнения.

fun main() {
    Observable.just("Task 1", "Task 2")
        .subscribeOn(Schedulers.single())
        .subscribe { println("Processing: $it on ${Thread.currentThread().name}") }
}

Запускает задачи в текущем потоке последовательно, помещая их в очередь. Новая задача выполняется только после завершения предыдущей. Это полезно для тестирования или строгой последовательности задач в одном потоке.

Observable.just("Task 1", "Task 2")
    .subscribeOn(Schedulers.trampoline())
    .subscribe { println("Processing: $it on ${Thread.currentThread().name}") }

Создаёт новый поток для каждой задачи. Используется, когда требуется изолированный поток выполнения, например, для независимых операций.

fun main() {
    Observable.just("Task 1", "Task 2")
        .subscribeOn(Schedulers.newThread())
        .subscribe { println("Processing: $it on ${Thread.currentThread().name}") }
}

Позволяет создавать кастомный планировщик (Scheduler) на основе переданного Executor. Это полезно, когда нужно контролировать управление потоками, например, задавать пул потоков.

fun main() {
    val executor = Executors.newFixedThreadPool(2)
    val scheduler = Schedulers.from(executor)

    Observable.just("Task 1", "Task 2")
        .subscribeOn(scheduler)
        .subscribe { println("Processing: $it on ${Thread.currentThread().name}") }

    executor.shutdown()
}

Возвращает Scheduler для выполнения операций в главном (UI) потоке Android. Используется, когда нужно обновить UI после выполнения асинхронных задач.

Observable.just("Hello, RxJava")
    .subscribeOn(Schedulers.io()) // Асинхронная задача выполняется в фоновом потоке
    .observeOn(AndroidSchedulers.mainThread()) // Результат возвращается в UI-поток
    .subscribe { value ->
        textView.text = value // Обновляем UI
    }

Отменяет подписку и освобождает ресурсы.

disposable.dispose()

Проверяет, была ли подписка отменена.

val isDisposed = disposable.isDisposed()

Позволяет управлять несколькими объектами Disposable одновременно. Он упрощает управление подписками, особенно когда у вас есть несколько потоков данных, на которые вы подписаны, и вы хотите отменить их все в одном месте.

fun main() {
    // Создание CompositeDisposable для управления подписками
    val compositeDisposable = CompositeDisposable()

    // Создание Observable
    val observable1 = Observable.just("Hello")
    val observable2 = Observable.just("World")

    // Подписка на первый Observable
    val disposable1: Disposable = observable1.subscribe(
        { item -> println(item) },
        { error -> println("Error: ${error.message}") },
        { println("Completed first observable") }
    )

    // Подписка на второй Observable
    val disposable2: Disposable = observable2.subscribe(
        { item -> println(item) },
        { error -> println("Error: ${error.message}") },
        { println("Completed second observable") }
    )

    // Добавление Disposables в CompositeDisposable
    compositeDisposable.addAll(disposable1, disposable2)

    // Отмена всех подписок
    compositeDisposable.dispose()

    // Проверка, были ли подписки отменены
    println("All disposed: ${compositeDisposable.size() == 0}")
}

Cтратегия по умолчанию, которая не применяет никакой политики к backpressure. Если источник данных генерирует события слишком быстро, возникает ошибка MissingBackpressureException.

Flowable.create<Int>({ emitter ->
    for (i in 1..1000) {
        emitter.onNext(i)
    }
    emitter.onComplete()
}, BackpressureStrategy.MISSING)
    .observeOn(Schedulers.io())
    .subscribe(
        { value -> println("Получено: $value") },
        { error -> println("Ошибка: $error") } // Вероятно, будет MissingBackpressureException
    )

Эта стратегия сохраняет все данные в буфере. Подходит, если данных не слишком много. Однако, если данных слишком много, может привести к переполнению памяти.

Flowable.create<Int>({ emitter ->
    for (i in 1..1000) {
        emitter.onNext(i)
    }
    emitter.onComplete()
}, BackpressureStrategy.BUFFER)
    .observeOn(Schedulers.io())
    .subscribe(
        { value -> println("Получено: $value") },
        { error -> println("Ошибка: $error") }
    )

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

Flowable.interval(1, TimeUnit.MILLISECONDS)
    .onBackpressureDrop()
    .observeOn(Schedulers.io())
    .subscribe(
        { value -> println("Получено: $value") },
        { error -> println("Ошибка: $error") }
    )

Сохраняет только последнее значение, отбрасывая остальные. Подходит, если важно всегда иметь доступ к самому последнему значению.

Flowable.interval(1, TimeUnit.MILLISECONDS)
    .onBackpressureLatest()
    .observeOn(Schedulers.io())
    .subscribe(
        { value -> println("Получено: $value") },
        { error -> println("Ошибка: $error") }
    )

Выбрасывает ошибку MissingBackpressureException, если данных слишком много. Используется, когда не нужно обрабатывать данные при перегрузке.

Flowable.create<Int>({ emitter ->
    for (i in 1..1000) {
        emitter.onNext(i)
    }
    emitter.onComplete()
}, BackpressureStrategy.ERROR)
    .observeOn(Schedulers.io())
    .subscribe(
        { value -> println("Получено: $value") },
        { error -> println("Ошибка: $error") } // Будет MissingBackpressureException
    )

Работает как и Observable, и Observer одновременно. Он начинает отправлять данные подписчикам только после их подписки, игнорируя все предыдущие события.

• Не сохраняет предыдущие данные. Новые подписчики получают только последующие события.

• Завершает поток с помощью onComplete().

• Может отправить ошибку через onError(). Если onError() был вызван, подписчики больше не будут получать данные.

val subject = PublishSubject.create<String>()

// Подписчик 1
subject.subscribe { value ->
    println("Подписчик 1: $value")
}

subject.onNext("Первое событие") // Подписчик 1 получит "Первое событие"

// Подписчик 2 подписывается после первого события
subject.subscribe { value ->
    println("Подписчик 2: $value")
}

subject.onNext("Второе событие") 
// Оба подписчика получат "Второе событие"