自己动手写一个RxJava框架？跟着这位博主，你也可以！

2023-09-22 07:39:05







## RxJava框架的基本概念

在开始实现RxJava框架之前，我们先来了解一下RxJava框架的基本概念。

* **被观察者（Observable）** ：被观察者是RxJava框架的核心概念之一，它代表了可以发射数据的源。被观察者可以是任何可以产生数据的对象，例如网络请求、传感器数据、用户输入等。
* **观察者（Observer）** ：观察者是另一个核心概念，它代表了对被观察者发射的数据感兴趣的对象。观察者可以是任何可以处理数据的对象，例如用户界面、日志记录器、其他RxJava操作符等。
* **订阅（Subscription）** ：订阅是将被观察者和观察者连接起来的桥梁。当观察者订阅被观察者时，它就会开始接收被观察者发射的数据。
* **发射器（Emitter）** ：发射器是RxJava框架中用于发射数据的对象。当被观察者需要发射数据时，它会通过发射器将数据发送给观察者。
* **订阅方法（subscribe）** ：subscribe方法是观察者订阅被观察者的方法。当观察者调用subscribe方法时，它就会开始接收被观察者发射的数据。
* **Disposable** ：Disposable是RxJava框架中用于取消订阅的对象。当观察者不再需要接收被观察者发射的数据时，它可以调用Disposable对象的dispose方法来取消订阅。

## 实现一个简化的RxJava框架

现在我们已经了解了RxJava框架的基本概念，接下来我们就来实现一个简化的RxJava框架。

### 1. 定义被观察者接口

首先，我们需要定义一个被观察者接口。这个接口定义了被观察者必须实现的方法，包括subscribe方法、onNext方法、onError方法和onComplete方法。

```java
public interface Observable<T> {

    void subscribe(Observer<T> observer);

    void onNext(T item);

    void onError(Throwable error);

    void onComplete();
}

2. 定义观察者接口

接下来，我们需要定义一个观察者接口。这个接口定义了观察者必须实现的方法，包括onNext方法、onError方法和onComplete方法。

public interface Observer<T> {

    void onNext(T item);

    void onError(Throwable error);

    void onComplete();
}

3. 定义订阅接口

接下来，我们需要定义一个订阅接口。这个接口定义了订阅必须实现的方法，包括dispose方法。

public interface Subscription {

    void dispose();
}

4. 定义发射器类

接下来，我们需要定义一个发射器类。这个类用于发射数据给观察者。

public class Emitter<T> {

    private Observer<T> observer;

    public Emitter(Observer<T> observer) {
        this.observer = observer;
    }

    public void onNext(T item) {
        observer.onNext(item);
    }

    public void onError(Throwable error) {
        observer.onError(error);
    }

    public void onComplete() {
        observer.onComplete();
    }
}

5. 定义订阅方法

接下来，我们需要定义订阅方法。这个方法用于将观察者订阅到被观察者。

public static <T> Subscription subscribe(Observable<T> observable, Observer<T> observer) {
    Emitter<T> emitter = new Emitter<>(observer);
    observable.subscribe(emitter);
    return emitter;
}

6. 定义create操作符

最后，我们需要定义create操作符。这个操作符用于创建被观察者。

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    return new Observable<T>() {

        @Override
        public void subscribe(Observer<T> observer) {
            source.subscribe(observer);
        }
    };
}

使用简化的RxJava框架

现在我们已经实现了一个简化的RxJava框架，接下来我们就来看看如何使用这个框架。

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {

    @Override
    public void subscribe(Observer<String> observer) {
        observer.onNext("Hello, world!");
        observer.onComplete();
    }
});

Observer<String> observer = new Observer<String>() {

    @Override
    public void onNext(String item) {
        System.out.println(item);
    }

    @Override
    public void onError(Throwable error) {
        System.out.println(error.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("Completed!");
    }
};

Subscription subscription = subscribe(observable, observer);

subscription.dispose();