问题标签 [rx-java2]

我正在使用 RxJava，并且Observable里面有多个项目。我想做的是在第一个项目上运行函数 A，在所有项目上运行函数 B，并在Observable完成时运行函数 C：

有没有用 lambda 函数表达这一点的聪明方法？我已经有以下解决方案，但它看起来很难看，我怀疑有更好的方法来做到这一点：

我想实现，如果我调用该Obervable.subscribe(Action1)方法，它不会OnErrorNotImplementedException在任何地方抛出，但如果我调用Obervable.subscribe(Action1, Action1)，则在正常引发错误时调用第二个操作。我尝试了两种方法：

这种方式OnErrorNotImplementedException不会被抛出，但是如果我也通过了第二个动作，那么这个动作也不会被调用。

第二：

如果稍后调用这个问题，observeOn()那么这将是异步的，显然我的异常处理在这里不起作用。

有没有办法做到这一点。我希望有一种subscribe()方法不会扔进OnErrorNotImplementedException去onError。

在 RxJava 1 中，有 CompositeSubscription，但在 RxJava2 中不存在，在 rxJava2 中有 CompositeDisposable。如何在 RxJava2 中使用 CompositeDisposable 或 Disposable？

我有 Observable 流，我想将其转换为 Completable，我该怎么做？

The doc says

Conceptually, it is a union of Single and Completable providing the means to capture an emission pattern where there could be 0 or 1 item or an error signalled by some reactive source.

But I am not sure what it truly means. It seems it is java8's Optional.

The following two codes have the same result , but I don't know what Maybe can do and Optional cannot (or cumbersome) do.

The results are the same :

In rxJava1 , My API used to return Observable<Optional<T>> , Is it a bad smell ? Should I change to Observable<Maybe<T>> ?

我的代码是：

那个代码对吗。PS：我使用的是 Android Studio 2，如何设置才能使用 lambda 表达式？

我的 RestClient 构造函数：

构建.gradle：

Lambda 语法现在应该可以通过添加 jackOptions 和 compileOptions 来工作

I'm looking to create a LocationHandler class that returns an observable<Location> whose I can send a new Location and subscribers get the last one added and any subsequent values.

I've written this class, it works but I don't know if it's the correct way to do it because I've added a callback and I smell it bad.

Thanks for any help.

Following @Blackbelt advice I've modified it with a ReplaySubject.

我有一个Observable<String>执行一些工作的。完成后，它会关闭其连接（通过.setDisposable(Disposables.fromAction(logic*);. 麻烦的是，我需要在与实际工作负载相同的调度程序上执行此关闭操作。

可能吗？如何？

我不想强制 Observable 在给定的调度程序上执行，例如 myObservable.subscribeOn(x).unsubscribeOn(x);；

当一些订阅者订阅了一个 flowable 并且这个 flowable 向订阅者发出项目时，每个订阅者可以得到不同的值吗？

例如，如果一个 flowable 发射

0, 1, 2, 3...

一个观察者得到 0、2，另一个观察者得到 1、3，依此类推，就像负载均衡器一样。

我有一个org.reactivestreams.Processor我想与 RxJava 2.0 一起使用的。然而，虽然有将 aorg.reactivestreams.Publisher与 RxJava 集成的转换，比如io.reactivex.Flowable#fromPublisher，但我不清楚如何最好地集成 a org.reactivestreams.Processor（或org.reactivestreams.Subscriber）。任何人都可以对此有所了解吗？