问题标签 [rx-java2]

以下代码

导致以下编译错误

错误：(9, 15) java: cannot access org.reactivestreams.Publisher class file for org.reactivestreams.Publisher not found

POM 依赖项如下

在 Hello World 示例中，有一个订阅者

如何制作两个或更多？

我在使用下面提到的库调用 API 时遇到问题

我的依赖：

提供改造实例的代码

API接口

最后是我调用 REST 的地方

结果我得到 404 - 当我在没有 RxJava 的情况下调用 API 时没有问题。有谁可以告诉我出了什么问题或为什么我会收到这样的错误？

onComplete如果一个可观察对象完成，我是否仍然需要取消订阅/处置（在 RxJava2 中）可观察对象以删除观察者（防止内存泄漏），或者一旦发生事件或事件，这是否由 RxJava 内部处理onError？

其他类型如Single,Completable等Flowable呢？

如何实现flatMapFirst类似的运算符flatMap，但只有在前一个结束时才添加新的 observable？如果前一个仍在运行，它只会忽略新的 observable。如何在 RxJava 2 中实现它？

它已经存在于 bacon - flatMapFirst 和 kefir - flatMapFirst中。

我刚刚开始学习如何在 Android 中使用 RxJava 2 和 Retrofit 2。我遵循了许多教程，并且能够从我的服务器中检索数据。我的问题是，轻松访问检索到的数据的标准方法是什么。目前 GSON 将其解析为一个对象，我打印 toString，但仅此而已。

我有这个接口来定义我的 POST 请求：

我创建改造对象：

然后我创建一个对象来包含我的 POST 请求的 JSON 正文：

然后我从我的界面创建 Observable：

最后，我订阅了 observable，在主线程上观察，但在工作线程上执行请求（我认为），然后记录结果：

toString 正确打印出从服务器接收到的用户信息。在 lambda 函数中使用这个对象的标准方法是什么？我想将它存储在数据库中，或者至少将它保存在活动中的对象中。我该怎么做呢？

我很难找到如何使用 RxJava 2 制作自定义运算符的示例。我考虑了一些方法：

1. 使用Observable.create, 然后flatMap从源 observable 对其进行 ing。我可以让它工作，但感觉不太对劲。我最终创建了一个提供源的静态函数Observable，然后在源上创建了 flatMap。在 OnSubscribe 中，然后我实例化一个我将发射器传递给的对象，该对象处理和管理 Observable / Emitter（因为它不是微不足道的，我希望一切都尽可能封装）。
2. 创建一个ObservableOperator并将其提供给Observable.lift. 对于 RxJava 2，我找不到任何这样的示例。我不得不调试自己的示例，以确保我对上游和下游的理解是正确的。因为我找不到任何关于 RxJava 2 的示例或文档，所以我有点担心我可能会不小心做一些我不应该做的事情。
3. 创建我自己的Observable类型。这似乎是底层运算符的工作方式，其中许多扩展了AbstractObservableWithUpstream. 不过这里发生了很多事情，似乎很容易错过一些事情或做一些我不应该做的事情。我不确定我是否应该采取这样的方法。我逐步完成了心理过程，似乎它很快就会变得毛茸茸。

我将继续使用选项#2，但认为值得询问 RxJava2 中支持的执行此操作的方法是什么，并找出是否有任何文档或示例。

我尝试在下面测试sereialize().

我调用onNext了 1,000,000 次以从 2 个不同的线程中计数。然后，我预计在onComplete.

但是，我无法获得预期值。

我想知道是否serialize()不起作用或我误解了的用法serialize()

我检查了 SerializedSubscriber 的来源。

由于actual.onNext(t);是从同步块中调用的，我想actual.onNext(t);可以同时从不同的线程调用。另外，我猜可能会在完成onComplete之前打电话。onNext

我使用了 RxJava 2.0.4。

这是我调用函数的代码

当值的长度很短时是正确的，当我将值的长度变长时，它会报错： java.lang.RuntimeException: Can't create handler inside thread that has not called Looper.prepare()

getData是一个抽象函数

在 RxJava1 中执行以下操作时，onNext 中的异常将被重新路由到同一个订阅者的 onError 中：

输出：

据我所知，在 RxJava2 中这样做时，这种行为已经改变，不再返回 onError，而是简单地崩溃：

输出：

我主要想知道，这两个版本中的哪个“错误”？这是已修复的 RxJava1 中的错误吗？这是 RxJava2 中的错误吗？或者这不是一个有意识的改变，因为我找不到任何细节？

附言。我确实注意到将其包装到“SafeObserver”中会再次重新路由到 onError