RxJava для Java: полное руководство по реактивному программированию

1. Что это и зачем нужна

RxJava — это библиотека для реактивного программирования на языке Java, реализующая концепцию Observable (наблюдаемого) и Observer (наблюдателя). Она позволяет обрабатывать асинхронные потоки данных и события с помощью композабельных (composable) операторов. В отличие от традиционных callback-ов или Future/Promise, RxJava предоставляет мощный набор инструментов для трансформации, фильтрации и комбинирования потоков данных, что делает код более читаемым, предсказуемым и менее подверженным ошибкам, связанным с многопоточностью.

Зачем нужна RxJava? В современной разработке Java-приложений (особенно Android, серверные приложения на Spring WebFlux, микросервисы) постоянно возникают задачи, связанные с асинхронностью: загрузка данных из сети, работа с базами данных, обработка пользовательских событий. RxJava решает эти задачи, предоставляя единый API для работы с любыми типами асинхронных последовательностей. Она позволяет избежать "callback hell" (ада из колбэков), упрощает обработку ошибок в асинхронных цепочках и предоставляет мощные средства для управления потоками (Schedulers).

Основные преимущества RxJava: декларативность (вы описываете, что нужно сделать, а не как), композабельность (операторы легко комбинируются), поддержка backpressure (противодавления) для управления скоростью потребления данных, и богатая экосистема (RxAndroid, RxJava2Extensions, RxRelay). Библиотека активно поддерживается и широко используется в индустрии, особенно в Android-разработке (Retrofit + RxJava, Room + RxJava).

2. Установка

RxJava распространяется через Maven Central. Для подключения к проекту добавьте зависимость в ваш файл build.gradle (для Gradle) или pom.xml (для Maven).

Gradle (Android / Java проект)

dependencies {
    implementation 'io.reactivex.rxjava3:rxjava:3.1.8'
}

Maven

<dependency>
    <groupId>io.reactivex.rxjava3</groupId>
    <artifactId>rxjava</artifactId>
    <version>3.1.8</version>
</dependency>

Для Android-проектов часто добавляют также RxAndroid для работы с главным потоком:

implementation 'io.reactivex.rxjava3:rxandroid:3.0.2'

3. Быстрый старт — минимальный рабочий пример

Создадим простой Observable, который генерирует строки "Hello" и "World", и подпишемся на него:

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;

public class RxJavaQuickStart {
    public static void main(String[] args) {
        // Создаем Observable, который испускает два элемента
        Observable<String> observable = Observable.just("Hello", "World");

        // Создаем Observer
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("Подписка установлена");
            }

            @Override
            public void onNext(String s) {
                System.out.println("Получено: " + s);
            }

            @Override
            public void onError(Throwable e) {
                System.err.println("Ошибка: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("Поток завершен");
            }
        };

        // Подписываемся
        observable.subscribe(observer);
    }
}

Вывод:

Подписка установлена
Получено: Hello
Получено: World
Поток завершен

4. Основные методы/классы

4.1 Observable.just()

Сигнатура: public static <T> Observable<T> just(T item1, T item2, ...)

Что делает: Создает Observable, который испускает переданные элементы последовательно и затем завершается.

Observable<Integer> numbers = Observable.just(1, 2, 3, 4, 5);
numbers.subscribe(System.out::println);
// Вывод: 1 2 3 4 5

4.2 Observable.fromIterable()

Сигнатура: public static <T> Observable<T> fromIterable(Iterable<? extends T> iterable)

Что делает: Создает Observable из любого Iterable (List, Set, Queue и т.д.).

List<String> list = Arrays.asList("A", "B", "C");
Observable<String> observable = Observable.fromIterable(list);
observable.subscribe(System.out::println);
// Вывод: A B C

4.3 Observable.map()

Сигнатура: public final <R> Observable<R> map(Function<? super T, ? extends R> mapper)

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

Observable<Integer> numbers = Observable.just(1, 2, 3);
Observable<String> mapped = numbers.map(n -> "Number: " + n);
mapped.subscribe(System.out::println);
// Вывод: Number: 1, Number: 2, Number: 3

4.4 Observable.filter()

Сигнатура: public final Observable<T> filter(Predicate<? super T> predicate)

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

Observable<Integer> numbers = Observable.just(1, 2, 3, 4, 5, 6);
Observable<Integer> evenNumbers = numbers.filter(n -> n % 2 == 0);
evenNumbers.subscribe(System.out::println);
// Вывод: 2 4 6

4.5 Observable.flatMap()

Сигнатура: public final <R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper)

Что делает: Преобразует каждый элемент в Observable и объединяет все полученные потоки в один (порядок не гарантируется).

Observable<String> words = Observable.just("Hello", "World");
Observable<String> letters = words.flatMap(word -> Observable.fromArray(word.split("")));
letters.subscribe(System.out::println);
// Вывод: H e l l o W o r l d (порядок может варьироваться)

4.6 Observable.concatMap()

Сигнатура: public final <R> Observable<R> concatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper)

Что делает: Аналогичен flatMap, но гарантирует сохранение порядка элементов (последовательное выполнение).

Observable<String> words = Observable.just("AB", "CD");
Observable<String> letters = words.concatMap(word -> Observable.fromArray(word.split("")));
letters.subscribe(System.out::println);
// Вывод: A B C D (всегда в правильном порядке)

4.7 Observable.zip()

Сигнатура: public static <T1, T2, R> Observable<R> zip(ObservableSource<? extends T1> source1, ObservableSource<? extends T2> source2, BiFunction<? super T1, ? super T2, ? extends R> zipper)

Что делает: Объединяет элементы из двух (или более) Observable по индексу с помощью функции-комбинатора.

Observable<Integer> numbers = Observable.just(1, 2, 3);
Observable<String> letters = Observable.just("A", "B", "C");
Observable<String> zipped = Observable.zip(numbers, letters, (n, l) -> n + l);
zipped.subscribe(System.out::println);
// Вывод: 1A, 2B, 3C

4.8 Observable.merge()

Сигнатура: public static <T> Observable<T> merge(ObservableSource<? extends T> source1, ObservableSource<? extends T> source2)

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

Observable<String> obs1 = Observable.just("A", "B");
Observable<String> obs2 = Observable.just("C", "D");
Observable<String> merged = Observable.merge(obs1, obs2);
merged.subscribe(System.out::println);
// Вывод: A B C D (порядок может быть любым)

4.9 Observable.subscribeOn() и Observable.observeOn()

Сигнатура: public final Observable<T> subscribeOn(Scheduler scheduler)
public final Observable<T> observeOn(Scheduler scheduler)

Что делает: subscribeOn определяет, на каком потоке будет выполняться Observable (создание данных). observeOn определяет, на каком потоке будут вызываться Observer (потребление данных).

import io.reactivex.rxjava3.schedulers.Schedulers;

Observable.just("Hello")
    .subscribeOn(Schedulers.io()) // Выполняем в IO-потоке
    .observeOn(Schedulers.single()) // Обрабатываем в single-потоке
    .subscribe(System.out::println);

4.10 Single, Maybe, Completable

Сигнатура: public abstract class Single<T>, public abstract class Maybe<T>, public abstract class Completable

Что делает: Специализированные типы Observable:
- Single — испускает ровно один элемент или ошибку.
- Maybe — испускает 0 или 1 элемент (или ошибку).
- Completable — только завершение или ошибка (без элементов).

// Single
Single.just("Result")
    .subscribe(System.out::println);

// Maybe
Maybe.empty()
    .subscribe(
        System.out::println,
        Throwable::printStackTrace,
        () -> System.out.println("Completed without value")
    );

// Completable
Completable.complete()
    .subscribe(() -> System.out.println("Done"));

5. Реальный практический пример

Пример: загрузка списка пользователей из API, их фильтрация по возрасту, преобразование в DTO и сохранение в базу данных с использованием RxJava.

import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.schedulers.Schedulers;
import java.util.*;
import java.util.concurrent.TimeUnit;

// Модели
class User {
    int id;
    String name;
    int age;
    User(int id, String name, int age) {
        this.id = id;
        this.name = name;
        this.age = age;
    }
}

class UserDTO {
    String displayName;
    boolean isAdult;
    UserDTO(String displayName, boolean isAdult) {
        this.displayName = displayName;
        this.isAdult = isAdult;
    }
}

// Репозиторий (имитация)
class UserRepository {
    public Single<List<User>> fetchUsers() {
        return Single.fromCallable(() -> {
            // Имитация задержки сети
            Thread.sleep(100);
            return Arrays.asList(
                new User(1, "Alice", 25),
                new User(2, "Bob", 17),
                new User(3, "Charlie", 30),
                new User(4, "David", 15)
            );
        }).subscribeOn(Schedulers.io());
    }

    public Completable saveUser(UserDTO dto) {
        return Completable.fromAction(() -> {
            // Имитация записи в БД
            System.out.println("Сохранен: " + dto.displayName);
            Thread.sleep(50);
        }).subscribeOn(Schedulers.io());
    }
}

public class RealWorldExample {
    public static void main(String[] args) throws InterruptedException {
        UserRepository repository = new UserRepository();

        // Цепочка обработки
        Disposable disposable = repository.fetchUsers()
            .flatMapObservable(Observable::fromIterable) // Преобразуем List в поток
            .filter(user -> user.age >= 18) // Фильтруем совершеннолетних
            .map(user -> new UserDTO(
                user.name.toUpperCase(),
                user.age >= 18
            ))
            .flatMapCompletable(dto -> repository.saveUser(dto)) // Сохраняем каждого
            .andThen(Single.just("Все пользователи обработаны"))
            .subscribe(
                result -> System.out.println("Результат: " + result),
                error -> System.err.println("Ошибка: " + error.getMessage()),
                () -> System.out.println("Завершено")
            );

        // Ждем завершения (в реальном приложении используем CompositeDisposable)
        Thread.sleep(2000);
    }
}

Вывод:

Сохранен: ALICE
Сохранен: CHARLIE
Результат: Все пользователи обработаны
Завершено

6. Советы по производительности и типичные ошибки

Советы по производительности:

Используйте правильный Scheduler:
- Schedulers.io() — для I/O операций (сеть, файлы, БД).
- Schedulers.computation()


                
    
  Также в библиотеке
  
                
                  
                 Работа с CSV в C++: библиотеки и примеры кода
      
                
                  
                 Работа с S3 и облачными хранилищами в Node.js: библиотека @aws-sdk/client-s3
      
                
                  
                 gRPC для Rust: библиотеки и примеры использования
      
                
                  
                 crossbeam для Rust: полное руководство с примерами
      
                
                  
                 JSON обработка в Java: библиотеки и инструменты для работы с JSON