RxJava2更新说明

朱雀 2022-10-30 09:24 147阅读 0赞

# Nulls #

RxJava 2.x 不再支持`null`值，以下写法会抛出`NullPointerException`:

Observable.just(null);
    
    Single.just(null);
    
    Observable.fromCallable(() -> null)
        .subscribe(System.out::println, Throwable::printStackTrace);
    
    Observable.just(1).map(v -> null)
        .subscribe(System.out::println, Throwable::printStackTrace);

# Observable 和 Flowable #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hndzEwMTA_size_16_color_FFFFFF_t_70]

*  背压  
    背压是指在异步场景中，被观察者发送事件速度远快于观察者的处理速度的情况下，一种告诉上游的被观察者降低发送速度的策略。

只有在需要处理背压问题时(10K+数据），才需要使用Flowable。

> 由于基于Flowable发射的数据流，以及对数据加工处理的各操作符都添加了背压支持，附加了额外的逻辑，其运行效率要比Observable慢得多。

# Single #

只接收一个事件。  
例子：

public void rxJavaSingleExample() { 
            Single.just(1)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new SingleObserver<Integer>() { 
                        @Override
                        public void onSubscribe(Disposable d) { 
    
                        }
    
                        @Override
                        public void onSuccess(Integer integer) { 
                            LogUtils.debug(TAG, "rxJavaSingleExample--:" + Thread.currentThread().getName() + "-onSuccess-:" + integer);
                        }
    
                        @Override
                        public void onError(Throwable e) { 
                            LogUtils.error(TAG, "rxJavaSingleExample--:" + Thread.currentThread().getName() + "-onError-:" + e.toString());
                        }
                    });
    
        }

# Completable #

在用法上与1.x基本一致。

只发射一条完成通知，或者一条异常通知，不能发射数据，其中完成通知与异常通知只能发射一个。

# Maybe #

功能上是`single`和`Completable`的结合。

可以发射一条单一的数据以及一条异常通知或者一条完成通知，需要注意的是，异常通知和完成通知只能选择其中一个，发射数据只能在完成通知或者异常通知之前，否则发送数据无效

# reactive 基本接口 #

参照Reactive-Streams的模式，`Flowable`里增加了`Publisher<T>`接口；`Observable`增加了`XSource`接口，如下：

interface ObservableSource<T> { 
        void subscribe(Observer<? super T> observer);
    }
    
    interface SingleSource<T> { 
        void subscribe(SingleObserver<? super T> observer);
    }
    
    interface CompletableSource { 
        void subscribe(CompletableObserver observer);
    }
    
    interface MaybeSource<T> { 
        void subscribe(MaybeObserver<? super T> observer);
    }

因此，许多操作新增支持`Publisher`和`XSource`类型，如：

Flowable<R> flatMap(Function<? super T, ? extends Publisher<? extends R>> mapper);
    
    Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper);

# Subjects 和 Processors #

*  Observable家族

io.reactivex.subjects.AsyncSubject  
io.reactivex.subjects.BehaviorSubject  
io.reactivex.subjects.PublishSubject  
io.reactivex.subjects.ReplaySubject  
io.reactivex.subjects.UnicastSubject

*  Flowable家族  
    io.reactivex.processors.AsyncProcessor  
    io.reactivex.processors.BehaviorProcessor  
    io.reactivex.processors.PublishProcessor  
    io.reactivex.processors.ReplayProcessor  
    io.reactivex.processors.UnicastProcessor

## TestSubject ##

更名为`TestScheduler`

TestScheduler scheduler = new TestScheduler();
    PublishSubject<Integer> ps = PublishSubject.create();
    
    TestObserver<Integer> ts = ps.delay(1000, TimeUnit.MILLISECONDS, scheduler)
    .test();
    
    ts.assertEmpty();
    
    ps.onNext(1);
    
    scheduler.advanceTimeBy(999, TimeUnit.MILLISECONDS);
    
    ts.assertEmpty();
    
    scheduler.advanceTimeBy(1, TimeUnit.MILLISECONDS);
    
    ts.assertValue(1);

## SerializedSubject ##

SerializedSubject不再能直接创建，需要调用`Subject.toSerialized()`或`FlowableProcessor.toSerialized()`。

# 其它类 #

`rx.observables.ConnectableObservable`改为`io.reactivex.observables.ConnectableObservable<T>`和`io.reactivex.flowables.ConnectableFlowable<T>`

## GroupedObservable ##

`rx.observables.GroupedObservable`改为`io.reactivex.observables.GroupedObservable<T>`和`io.reactivex.flowables.GroupedFlowable<T>`，并且是抽象类。使用时需要自己实现对应的方法：

class MyGroup<K, V> extends GroupedObservable<K, V> { 
        final K key;
    
        final Subject<V> subject;
    
        public MyGroup(K key) { 
            this.key = key;
            this.subject = PublishSubject.create();
        }
    
        @Override
        public T getKey() { 
            return key;
        }
    
        @Override
        protected void subscribeActual(Observer<? super T> observer) { 
            subject.subscribe(observer);
        }
    }

`GroupedFlowable`也一样。

# 函数接口 #

添加了异常抛出的处理：

Flowable.just("file.txt")
    .map(name -> Files.readLines(name))
    .subscribe(lines -> System.out.println(lines.size()), Throwable::printStackTrace);

当文件不存在或读取出错时，消费者会打印`IOException`。注意`Files.readLines(name)`不要添加 try-catch。

## Actions ##

*  无参的`Action0`替换为`io.reactivex.functions.Action`
 *  `Action1`重命名为`Consumer`
 *  `Action2`重命名为`BiConsumer`
 *  `ActionN` 替换为`Consumer<Object[]>`
 *  删除`Action3`\-`Action9`

## Functions ##

*  增加`io.reactivex.functions.Function`和`io.reactivex.functions.BiFunction`
 *  `Func3` \- `Func9`重命名为`Function3` \- `Function9`
 *  `FuncN`替换为`Function<Object[], R>`
 *  需要predicate的操作不再使用`Func1<T, Boolean>`，而是使用`Predicate<T>`

# Subscriber #

在onSubscribe/onStart中调用request方法，会阻塞直至事件发送完。

Flowable.range(1, 3).subscribe(new Subscriber<Integer>() { 
    
        @Override
        public void onSubscribe(Subscription s) { 
            System.out.println("OnSubscribe start");
            s.request(Long.MAX_VALUE);
            System.out.println("OnSubscribe end");
        }
    
        @Override
        public void onNext(Integer v) { 
            System.out.println(v);
        }
    
        @Override
        public void onError(Throwable e) { 
            e.printStackTrace();
        }
    
        @Override
        public void onComplete() { 
            System.out.println("Done");
        }
    });

输出：

OnSubscribe start
    1
    2
    3
    Done
    OnSubscribe end

# Subscription #

在2.x中，`subscribe`方法返回`Disposable`对象。

`Subscription`类型被重命名：

*  `CompositeSubscription`改为`CompositeDisposable`
 *  `SerialSubscription`和`MultipleAssignmentSubscription`合并为`SerialDisposable`
 *  删除`RefCountSubscription`

# 异常处理 #

全局异常处理：

RxJavaPlugins.setErrorHandler(e -> { 
        if (e instanceof UndeliverableException) { 
            e = e.getCause();
        }
        if ((e instanceof IOException) || (e instanceof SocketException)) { 
            // fine, irrelevant network problem or API that throws on cancellation
            return;
        }
        if (e instanceof InterruptedException) { 
            // fine, some blocking code was interrupted by a dispose call
            return;
        }
        if ((e instanceof NullPointerException) || (e instanceof IllegalArgumentException)) { 
            // that's likely a bug in the application Thread.currentThread().getUncaughtExceptionHandler() .handleException(Thread.currentThread(), e); return; } if (e instanceof IllegalStateException) { // that's a bug in RxJava or in a custom operator
            Thread.currentThread().getUncaughtExceptionHandler()
                .handleException(Thread.currentThread(), e);
            return;
        }
        Log.warning("Undeliverable exception received, not sure what to do", e);
    });

参考：  
https://github.com/ReactiveX/RxJava/wiki/What’s-different-in-2.0  
https://www.cnblogs.com/shwo/p/9874680.html  
https://www.jianshu.com/p/0cd258eecf60

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hndzEwMTA_size_16_color_FFFFFF_t_70]: /images/20221024/b499501e6d5a42a3aebe810ebce592f2.png