rxjava怎么创建观察者

在RxJava中，创建观察者通常涉及以下几个步骤：

1. 定义数据源：首先，你需要有一个数据源，这可以是一个Observable对象，它负责发出数据项。

2. 创建观察者：然后，你需要创建一个Observer对象，它将订阅数据源并处理发出的数据项。

3. 订阅：最后，你需要调用Observable的subscribe()方法来建立数据流，并将观察者注册到数据源上。

下面是一个简单的例子，展示了如何使用RxJava 2.x创建一个观察者：

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

public class RxJavaExample {
    public static void main(String[] args) {
        // 创建一个Observable对象，它将发出三个字符串数据项
        Observable<String> observable = Observable.just("Hello", "RxJava", "Observer");

        // 创建一个Observer对象
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                // 当订阅建立时调用，可以在这里处理订阅相关的逻辑
                System.out.println("Subscribed");
            }

            @Override
            public void onNext(String s) {
                // 当Observable发出数据项时调用
                System.out.println("Received: " + s);
            }

            @Override
            public void onError(Throwable e) {
                // 当Observable发生错误时调用
                System.err.println("Error: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                // 当Observable完成数据发射时调用
                System.out.println("Completed");
            }
        };

        // 订阅Observable，将观察者注册到数据源上
        observable.subscribe(observer);
    }
}

在RxJava 3.x中，创建观察者的方式略有不同，使用的是Observer接口的新版本，以及Disposable接口的新版本。以下是RxJava 3.x的等效代码：

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

public class RxJava3Example {
    public static void main(String[] args) {
        // 创建一个Observable对象，它将发出三个字符串数据项
        Observable<String> observable = Observable.just("Hello", "RxJava", "Observer");

        // 创建一个Observer对象
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                // 当订阅建立时调用
                System.out.println("Subscribed");
            }

            @Override
            public void onNext(String s) {
                // 当Observable发出数据项时调用
                System.out.println("Received: " + s);
            }

            @Override
            public void onError(Throwable e) {
                // 当Observable发生错误时调用
                System.err.println("Error: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                // 当Observable完成数据发射时调用
                System.out.println("Completed");
            }
        };

        // 订阅Observable，将观察者注册到数据源上
        observable.subscribe(observer);
    }
}

在这两个例子中，我们创建了一个简单的Observable，它发出了三个字符串数据项。然后，我们创建了一个Observer来订阅这个Observable，并实现了onSubscribe、onNext、onError和onComplete方法来处理不同的事件。最后，我们通过调用subscribe()方法将观察者注册到Observable上，从而建立了数据流。