您现在的位置是：首页 > Java

当前栏目

RxJava之创建操作符源码介绍

2023-02-18 16:34:07 时间

转载请以链接形式标明出处：本文出自:103style的博客

前言

前置阅读：RxJava之create操作符源码解析

Rxjava 之 创建操作符 官方介绍：Creating-Observables

创建相关的操作符以及官方介绍

前置阅读：Rxjava之create操作符源码解析，已经很详细的介绍了 create操作符，如果你还没有阅读过，请先阅读 Rxjava之create操作符源码解析。

创建相关的操作符源码介绍

为了方便介绍，顺序有变化

`create` 操作符

Rxjava之create操作符源码解析

`just` 操作符

官方提供的使用例子：

String greeting = "Hello world!";
Observable<String> observable = Observable.just(greeting);
observable.subscribe(item -> System.out.println(item));

just操作符实际上返回的是一个 ObservableJust对象，just 操作符也提供了2-10个参数的，不过内部是调用了fromArray 操作符。

public static <T> Observable<T> just(T item) {
    return RxJavaPlugins.onAssembly(new ObservableJust<T>(item));
}

public static <T> Observable<T> just(T item1, T item2) {
    return fromArray(item1, item2);
}
...

ObservableJust源码：

public final class ObservableJust<T> extends Observable<T> implements ScalarCallable<T> {

    private final T value;
    public ObservableJust(final T value) {
        this.value = value;
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        ScalarDisposable<T> sd = new ScalarDisposable<T>(observer, value);
        observer.onSubscribe(sd);
        sd.run();
    }

    @Override
    public T call() {
        return value;
    }
}

主要看subscribeActual方法，我们看到，

首先构建了一个ScalarDisposable对象，
然后调用 观察者 的 onSubscribe方法，
接着执行 ScalarDisposable 的 run方法。

ScalarDisposable 相关源码：

public static final class ScalarDisposable<T>
extends AtomicInteger
implements QueueDisposable<T>, Runnable {
    static final int START = 0;
    static final int FUSED = 1;
    static final int ON_NEXT = 2;
    static final int ON_COMPLETE = 3;
  ...
    @Override
    public void run() {
        if (get() == START && compareAndSet(START, ON_NEXT)) {
            observer.onNext(value);
            if (get() == ON_NEXT) {
                lazySet(ON_COMPLETE);
                observer.onComplete();
            }
        }
    }
}

run方法里面我们又看到了之前 Rxjava之create操作符源码解析里介绍的 get()和compareAndSet(int var1, int var2)方法。

因为是AtomicInteger，所以 get() 获取到的 value 值为 int 类型的默认值 0，0则为 ObservableJust 的 START 状态。
compareAndSet(int var1, int var2) 是比较 var1 和 var2 的值，如果不相等，则更新 var1 的值为var2，并返回true。所以 ObservableJust 由 START 进入ON_NEXT 状态。
run()方法的 if 条件 get() == START && compareAndSet(START, ON_NEXT) 成立。所以将 just操作符传进来的value值传入 观察者 的 onNext 方法
因为此时为onNext状态，所以 get() == ON_NEXT 成立，修改 ObservableJust 的状态为 onComplete，继续执行 观察者 的 onComplete 方法。

当然我们也可以在 观察者 的 onSubscribe方法，调用 dispose() 修改 ObservableJust 的状态。

public static final class ScalarDisposable<T>
extends AtomicInteger
implements QueueDisposable<T>, Runnable {
    ...        
    @Override
    public void dispose() {
        set(ON_COMPLETE);
    }
}

`defer` 操作符

官方提供的使用例子：

Observable<Long> observable = Observable.defer(() -> {
    long time = System.currentTimeMillis();
    return Observable.just(time);
});

observable.subscribe(time -> System.out.println(time));

Thread.sleep(1000);

observable.subscribe(time -> System.out.println(time));

defer操作符实际上返回的是一个 ObservableDefer对象。

public static <T> Observable<T> defer(Callable<? extends ObservableSource<? extends T>> supplier) {
    ObjectHelper.requireNonNull(supplier, "supplier is null");
    return RxJavaPlugins.onAssembly(new ObservableDefer<T>(supplier));
}

ObservableDefer 源码：

public final class ObservableDefer<T> extends Observable<T> {
    final Callable<? extends ObservableSource<? extends T>> supplier;

    public ObservableDefer(Callable<? extends ObservableSource<? extends T>> supplier) {
        this.supplier = supplier;
    }

    @Override
    public void subscribeActual(Observer<? super T> observer) {
        ObservableSource<? extends T> pub;
        try {
            pub = ObjectHelper.requireNonNull(supplier.call(), "null ObservableSource supplied");
        } catch (Throwable t) {
            Exceptions.throwIfFatal(t);
            EmptyDisposable.error(t, observer);
            return;
        }
        pub.subscribe(observer);
    }
}

defer操作符实际上是将 call() 方法返回的 ObservableSource 和 subscribe 中的 观察者 建立订阅关系。

Observable.defer(new Callable<ObservableSource<Object>>() {
    @Override
    public ObservableSource<Object> call() throws Exception {
        return null;
    }
}).subscribe(new Observer<Object>() {...});

官方的例子中第一次调用 subscribe 之后，延迟一秒之后再次调用 subscribe，相当与调用 call() 之后，延迟一秒之后再次调用 call()，即执行两次Observable.just(当前毫秒数)。看了上面的 just 操作符介绍，我们可以得知，会执行两次 subscribe传进来的 Observer 的 onNext(Object o)方法。

`empty` 操作符

empty操作符实际上返回的是一个 ObservableEmpty对象。

public static <T> Observable<T> empty() {
    return RxJavaPlugins.onAssembly((Observable<T>) ObservableEmpty.INSTANCE);
}

ObservableEmpty源码：

public final class ObservableEmpty extends Observable<Object> implements ScalarCallable<Object> {
    public static final Observable<Object> INSTANCE = new ObservableEmpty();

    private ObservableEmpty() {
    }

    @Override
    protected void subscribeActual(Observer<? super Object> o) {
        EmptyDisposable.complete(o);
    }

    @Override
    public Object call() {
        return null; 
    }
}

EmptyDisposable 的 complete(Observer<?> observer) 方法

public enum EmptyDisposable implements QueueDisposable<Object> {
    ...
    public static void complete(Observer<?> observer) {
        observer.onSubscribe(INSTANCE);
        observer.onComplete();
    }
    ...
}

所以我们看到 empty 操作符实际上，在订阅之后直接执行了 观察者 的 onSubscribe(Disposable d) 和 onComplete() 方法。

`error` 操作符

error 操作符实际上返回的是一个 ObservableError对象。和 empty 操作符类似，在订阅之后直接执行了 观察者 的 onSubscribe(Disposable d) 和 onError(Throwable e) 方法。

public static <T> Observable<T> error(Callable<? extends Throwable> errorSupplier) {
    ObjectHelper.requireNonNull(errorSupplier, "errorSupplier is null");
    return RxJavaPlugins.onAssembly(new ObservableError<T>(errorSupplier));
}

public final class ObservableError<T> extends Observable<T> {
    final Callable<? extends Throwable> errorSupplier;
    public ObservableError(Callable<? extends Throwable> errorSupplier) {
        this.errorSupplier = errorSupplier;
    }

    @Override
    public void subscribeActual(Observer<? super T> observer) {
        ...
        EmptyDisposable.error(error, observer);
    }
}

public enum EmptyDisposable implements QueueDisposable<Object> {
    ...
    public static void error(Throwable e, Observer<?> observer) {
        observer.onSubscribe(INSTANCE);
        observer.onError(e);
    }
    ...
}

`from` 操作符系列

Rxjava之from系列操作符源码解析

`generate` 操作符

generate 操作符实际上返回的是一个 ObservableGenerate对象。

首先获取 Callable 对象 stateSupplier 的 call 方法的返回值。

构建 GeneratorDisposable 对象。

然后调用 观察者 的 onSubscribe(Disposable d)。

执行GeneratorDisposable 对象的 run 方法。

public final class ObservableGenerate<T, S> extends Observable<T> {
    final Callable<S> stateSupplier;
    final BiFunction<S, Emitter<T>, S> generator;
    final Consumer<? super S> disposeState;

    public ObservableGenerate(Callable<S> stateSupplier, BiFunction<S, Emitter<T>, S> generator,
            Consumer<? super S> disposeState) {
        this.stateSupplier = stateSupplier;
        this.generator = generator;
        this.disposeState = disposeState;
    }

    @Override
    public void subscribeActual(Observer<? super T> observer) {
        S state;
        try {
            state = stateSupplier.call();
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            EmptyDisposable.error(e, observer);
            return;
        }
        GeneratorDisposable<T, S> gd = new GeneratorDisposable<T, S>(observer, generator, disposeState, state);
        observer.onSubscribe(gd);
        gd.run();
    }
    ...
}

GeneratorDisposable 源码：

run 方法里面无限循环执行 generate操作符传进来的 Consumer对象的 accept 方法。直到在 accept 中调用 onComplete() 或者 onError方法
在Consumer对象的 accept 方法执行 onNext 则会调用观察者的 onNext 。

static final class GeneratorDisposable<T, S>
implements Emitter<T>, Disposable {
    final Observer<? super T> downstream;
    final BiFunction<S, ? super Emitter<T>, S> generator;
    final Consumer<? super S> disposeState;
    S state;
    volatile boolean cancelled;
    boolean terminate;
    boolean hasNext;

    GeneratorDisposable(Observer<? super T> actual,
            BiFunction<S, ? super Emitter<T>, S> generator,
            Consumer<? super S> disposeState, S initialState) {
        this.downstream = actual;
        this.generator = generator;
        this.disposeState = disposeState;
        this.state = initialState;
    }

    public void run() {
        S s = state;
        if (cancelled) {
            state = null;
            dispose(s);
            return;
        }
        final BiFunction<S, ? super Emitter<T>, S> f = generator;
        for (;;) {
            if (cancelled) {
                state = null;
                dispose(s);
                return;
            }
            hasNext = false;
            try {
                s = f.apply(s, this);
            } catch (Throwable ex) {
                ...
                onError(ex);
                dispose(s);
                return;
            }
            if (terminate) {
                cancelled = true;
                state = null;
                dispose(s);
                return;
            }
        }
    }
    ...
}

单参数默认构建的 SimpleGenerator:

static final class SimpleGenerator<T, S> implements BiFunction<S, Emitter<T>, S> {
    final Consumer<Emitter<T>> consumer;

    SimpleGenerator(Consumer<Emitter<T>> consumer) {
        this.consumer = consumer;
    }

    @Override
    public S apply(S t1, Emitter<T> t2) throws Exception {
        consumer.accept(t2);
        return t1;
    }
}

`never` 操作符

never 操作符实际上返回的是一个 ObservableNever对象。只执行 观察者 的 onSubscribe方法。

public final class ObservableNever extends Observable<Object> {
    public static final Observable<Object> INSTANCE = new ObservableNever();

    private ObservableNever() {
    }

    @Override
    protected void subscribeActual(Observer<? super Object> o) {
        o.onSubscribe(EmptyDisposable.NEVER);
    }
}

`range` 操作符 `range(final int start, final int count)`

count 参数为 0 时执行 empty 操作符，count 为 1 时执行 just 操作符。

range 操作符实际上返回的是一个 ObservableRange对象。

首先初始化了 起始点 和 结束点 的值。
接着构建 RangeDisposable 对象。
接着调用 观察者 的 onSubscribe(Disposable d)方法。
然后执行 RangeDisposable 对象的 run 方法。

 public final class ObservableRange extends Observable<Integer> {
     private final int start;
     private final long end;
 
     public ObservableRange(int start, int count) {
         this.start = start;
         this.end = (long)start + count;
     }
 
     @Override
     protected void subscribeActual(Observer<? super Integer> o) {
         RangeDisposable parent = new RangeDisposable(o, start, end);
         o.onSubscribe(parent);
         parent.run();
     }
 }

RangeDisposable 对象的 run方法：

从 起始点 到 结束点 的循环遍历，状态对的话则调用 观察者 的 onNext 方法。

static final class RangeDisposable
extends BasicIntQueueDisposable<Integer> {
    final Observer<? super Integer> downstream;
    final long end;
    long index;
    boolean fused;

    RangeDisposable(Observer<? super Integer> actual, long start, long end) {
        this.downstream = actual;
        this.index = start;
        this.end = end;
    }

    void run() {
        if (fused) {
            return;
        }
        Observer<? super Integer> actual = this.downstream;
        long e = end;
        for (long i = index; i != e && get() == 0; i++) {
            actual.onNext((int)i);
        }
        if (get() == 0) {
            lazySet(1);
            actual.onComplete();
        }
    }
    ...

`interval` 和 `timer` 操作符

Rxjava之timer和interval操作符源码解析

以上

猜你喜欢

C#.NET 大型通用信息化系统集成快速开发平台 4.0 版本 - 组织机构的名称编号是否允许重复？
C#.NET 大型通用信息化系统集成快速开发平台 4.0 版本 - 导入导出Microsoft Excel 2010的例子
C#.NET 大型通用信息化系统集成快速开发平台 4.0 版本 - 省市区数据权限的实现效果
C#.NET 大型通用信息化系统集成快速开发平台 4.0 版本 - 标准省市县数据的公司选择窗口实现
C#.NET 大型通用信息化系统集成快速开发平台 4.0 版本 - 独立子系统管理员功能实现
通用用户权限管理系统组件4.0 版本 - 界面输入安全检查 - 为大规模开发软件系统做样例
K8 系统中省市县数据表的设计可以反映出什么？通过一个基础业务表的设计品味软件系统的整体架构
C#.NET 大型通用信息化系统集成快速开发平台 4.0 版本 - 用户权限树的实现 -- 权限递归树
C#.NET 大型通用信息化系统集成快速开发平台 4.0 版本 - 多系统开发接口 - 苹果客户端开发接口
大数据下载功能优化实现
2013圣诞节前程序员对话记录留念
通用权限管理系统组件-数据权限参考
经得起严酷实战考验的高性能到大型信息系统的核心底层-通用权限管理系统组件
C#开发中Windows域认证登录2（扩展吉日嘎拉GPM系统）
C#.NET 大型信息化系统集成快速开发平台 - 手机短信开发接口 4.0
多软件子系统集成的效果图-通用权限管理系统组件4.0版本
通用权限管理系统组件回答用户的常用问题：操作权限、用户角色、数据权限的解决方法
通用权限管理系统多语言开发标准接口 - java，php 调用标准接口程序参考
一起打造.NET职业生涯奇迹中国十大快递公司之一的核心业务系统采用.NET通用权限管理系统组件架构重构［重磅博客园独有福利］
通用权限管理系统组件3.9 的 Oracle 数据库创建脚本参考

zl程序教程

当前栏目

RxJava之创建操作符源码介绍

前言

创建相关的操作符以及官方介绍

创建相关的操作符源码介绍

`create` 操作符

`just` 操作符

`defer` 操作符

`empty` 操作符

`error` 操作符

`from` 操作符系列

`generate` 操作符

`never` 操作符

`range` 操作符 `range(final int start, final int count)`

`interval` 和 `timer` 操作符

相关文章

当前栏目

RxJava之创建操作符源码介绍

前言

创建相关的操作符 以及 官方介绍

创建相关的操作符源码介绍

create 操作符

just 操作符

defer 操作符

empty 操作符

error 操作符

from 操作符系列

generate 操作符

never 操作符

range 操作符 range(final int start, final int count)

interval 和 timer 操作符

相关文章

创建相关的操作符以及官方介绍

`create` 操作符

`just` 操作符

`defer` 操作符

`empty` 操作符

`error` 操作符

`from` 操作符系列

`generate` 操作符

`never` 操作符

`range` 操作符 `range(final int start, final int count)`

`interval` 和 `timer` 操作符