标签: rx-java2

我没有理由在Android中使用RxJava,在Android Architectural Components中使用LiveData.如果两者之间的用例和差异以及代码形式的示例示例进行解释,这将解释两者之间的差异,这将非常有用.

我今天已升级到Android Studio 3.1,这似乎增加了一些lint检查.其中一个lint检查是针对subscribe()未存储在变量中的一次性RxJava2 调用.例如,从我的Room数据库中获取所有玩家的列表:

Single.just(db)
            .subscribeOn(Schedulers.io())
            .subscribe(db -> db.playerDao().getAll());

导致一个大的黄色块和这个工具提示:

结果subscribe未使用

这样的一次性Rx调用的最佳做法是什么？我应该保持的保持Disposable,并dispose()在完成了吗？或者我应该@SuppressLint继续前进？

编辑这似乎只影响RxJava2(io.reactivex),RxJava(rx)没有这个lint.

任何人都可以用明确的例子解释RxJava中Observable,Completable和Single之间的区别吗？

在哪种情况下我们使用其中一个？

我将向我的服务器(它是Rails应用程序)发送一个简单的get方法,并使用RxJava和Retrofit获取结果.我做的是:

我的界面:

public interface ApiCall {
    String SERVICE_ENDPOINT = "https://198.50.214.15";
    @GET("/api/post")
    io.reactivex.Observable<Post> getPost();
}

我的模型是这样的:

public class Post
{
    @SerializedName("id")
    private String id;
    @SerializedName("body")
    private String body;
    @SerializedName("title")
    private String title;

    public String getId ()
    {
        return id;
    }


    public String getBody ()
    {
        return body;
    }


    public String getTitle ()
    {
        return title;
    }

}

这就是我在活动中所做的:

public class Javax extends AppCompatActivity {
    RecyclerView rvListContainer;
    postAdapter postAdapter;
    List<String> messageList=new ArrayList<>();
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_javax);

        rvListContainer=(RecyclerView)findViewById(R.id.recyclerView);
        postAdapter=new …

在RxJava 1中,有CompositeSubscription,但在RxJava2中不存在,rxJava2中有一些CompositeDisposable.如何在RxJava2中使用CompositeDisposable或Disposable？

在最近几个主要版本的Java的每次迭代中,似乎都有新的方法来管理并发任务.

在Java 9中,我们有Flow API,类似于RxJava的Flowable API,但Java 9有一组更简单的类和接口.

Java 9

有Flow.Publisher,Flow.Subscriber,Flow.Processor,Flow.Subscription,和SubmissionPublisher,这就是它.

RxJava

拥有全包的流API状类,即io.reactivex.flowables,io.reactivex.subscribers,io.reactivex.processors,io.reactivex.observers,和io.reactivex.observables这似乎做同样的事情.

这两个库之间的主要区别是什么？为什么有人会使用Java 9 Flow库而不是更多样化的RxJava库,反之亦然？

我在尝试为正在使用的演示者运行JUnit测试时遇到RuntimeException observeOn(AndroidSchedulers.mainThread()).

由于它们是纯JUnit测试而不是Android检测测试,因此它们无法访问Android依赖项,导致我在执行测试时遇到以下错误:

java.lang.ExceptionInInitializerError
    at io.reactivex.android.schedulers.AndroidSchedulers$1.call(AndroidSchedulers.java:35)
    at io.reactivex.android.schedulers.AndroidSchedulers$1.call(AndroidSchedulers.java:33)
    at io.reactivex.android.plugins.RxAndroidPlugins.callRequireNonNull(RxAndroidPlugins.java:70)
    at io.reactivex.android.plugins.RxAndroidPlugins.initMainThreadScheduler(RxAndroidPlugins.java:40)
    at io.reactivex.android.schedulers.AndroidSchedulers.<clinit>(AndroidSchedulers.java:32)
    …
Caused by: java.lang.RuntimeException: Method getMainLooper in android.os.Looper not mocked. See http://g.co/androidstudio/not-mocked for details.
    at android.os.Looper.getMainLooper(Looper.java)
    at io.reactivex.android.schedulers.AndroidSchedulers$MainHolder.<clinit>(AndroidSchedulers.java:29)
    ...


java.lang.NoClassDefFoundError: Could not initialize class io.reactivex.android.schedulers.AndroidSchedulers
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    …

何时调用subscribeWith方法而不是普通订阅？什么是用例？

compositeDisposable.add(get()
    .observeOn(AndroidSchedulers.mainThread())
    .subscribeOn(Schedulers.io())
    .subscribe(this::handleResponse, this::handleError));

VS

   compositeDisposable.add(get()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.io())
              //  .subscribe(this::handleResponse, this::handleError);
                .subscribeWith(new DisposableObserver<News>() {
                    @Override public void onNext(News value) {
                        handleResponse(value);
                    }

                    @Override public void onError(Throwable e) {
                        handleError(e);
                    }

                    @Override public void onComplete() {
                       // dispose here ? why? when the whole thing will get disposed later
                       //via  compositeDisposable.dispose();  in onDestroy();
                    }
                }));

谢谢

稍后添加

根据文档,两者都返回一次性SingleObserver实例:

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <E extends SingleObserver<? super T>> E subscribeWith(E observer) {
    subscribe(observer);
    return observer;
}

@SchedulerSupport(SchedulerSupport.NONE)
public final Disposable …

我有两个可完成的.我想做以下场景:如果第一个Completable到达onComplete,继续第二个Completable.最终结果将是第二次完成的完成.

当我有单个getUserIdAlreadySavedInDevice()和Completable login()时,我就是这样做的:

@Override
public Completable loginUserThatIsAlreadySavedInDevice(String password) {
    return getUserIdAlreadySavedInDevice()
            .flatMapCompletable(s -> login(password, s))

}

据我所知,RxJava2 values.take(1)创建了另一个Observable,它只包含原始Observable中的一个元素.哪个不能抛出异常,因为它被take(1)第二次发生的效果过滤掉了.

如下面的代码片段所示

    Observable<Integer> values = Observable.create(o -> {
        o.onNext(1);
        o.onError(new Exception("Oops"));
    });

    values.take(1)
            .subscribe(
                    System.out::println,
                    e -> System.out.println("Error: " + e.getMessage()),
                    () -> System.out.println("Completed")
            );

产量

1
Completed
io.reactivex.exceptions.UndeliverableException: java.lang.Exception: Oops
    at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:366)
    at io.reactivex.internal.operators.observable.ObservableCreate$CreateEmitter.onError(ObservableCreate.java:83)
    at ch02.lambda$main$0(ch02.java:28)
    at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
    at io.reactivex.Observable.subscribe(Observable.java:10841)
    at io.reactivex.internal.operators.observable.ObservableTake.subscribeActual(ObservableTake.java:30)
    at io.reactivex.Observable.subscribe(Observable.java:10841)
    at io.reactivex.Observable.subscribe(Observable.java:10827)
    at io.reactivex.Observable.subscribe(Observable.java:10787)
    at ch02.main(ch02.java:32)
Caused by: java.lang.Exception: Oops
    ... 8 more
Exception in thread "main" io.reactivex.exceptions.UndeliverableException: java.lang.Exception: Oops
    at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:366)
    at io.reactivex.internal.operators.observable.ObservableCreate$CreateEmitter.onError(ObservableCreate.java:83)
    at ch02.lambda$main$0(ch02.java:28)
    at …