问题标签 [rx-java3]

我正在从文件中读取内容，我用换行符将一些名称放入文件中，但是在通过 vertx 文件系统读取文件时，我无法提供所需的过滤器。每次打印文件的所有数据。

这是代码片段：-

在我发现上述 observable 一口气拥有文件的所有数据后，我使用了内部 observable。

为什么subscribe这里从不打印任何东西？只是出于好奇。无论如何，这都是不好的做法：我通常会使用它observeOn。但是，我无法弄清楚为什么subscribe从未达到...

使用调度程序进行测试的 RxJava 规则是

现在使用 RxJava3

这里调用了第二个参数，fair但类中没有 JavaDoc 或该参数的解释ExecutorScheduler。公平参数的用途是什么，应该在什么时候使用？

我正在尝试将我的前端代码转换为 java 本机代码。但是如何转换以及我可以在 rxjava3 中使用的类似运算符有点困难。

Forntend rxjs 示例

与上面相同，我想在使用 rxjava3 的 java 1.8 程序中实现。如何？

任何帮助都非常受欢迎。

谢谢。

我正在使用 RxJava2、Micronaut 和 Cassandra 处理响应式流。我是 rxjava 的新手，不确定以最佳异步方式返回 List Person 的正确方法是什么？

数据来自 Cassandra Dao 接口

注入到 micronaut 控制器中

或者

或切换到 RxJava3

我正在使用rxjava3并且不太理解为什么添加ObserveOnand后方法没有在流中被调用SubscribeOn。

这是示例 java 代码：

我想在单独的线程上执行每个方法，也想在不同的线程上执行订阅。

造成问题的原因是什么。如何解决上述情况？

谢谢。

我偶然发现了一个奇怪的 testScheduler 行为，我无法理解。下面的代码被大大简化了，但它源于现实生活中的问题。

考虑这个测试：

**当我订阅 test stringsProcessorontestScheduler时，测试失败。当我删除该行时，它成功了。**

我看到的事件流程是：

1. 触发动作
2. 正在订阅 completable 和 stringsProcessor 并将它们的事件传播到下游。
3. 显然，在 testSubscriber 完成后stringsProcessor.onNext("onComplete")评估。

我想知道为什么

我是 RxAndroid 的新手，这就是为什么提到 Rx 的所有运营商。我发现它observable.takeFirst()正在显示红线，因为它在 RxJava3 中找不到。谁能告诉我方法的替代takefirst()方法是什么？

在下面的代码行中

我希望日志应该说“12345”，但它说的是 23415、31245、53124 等。（即订单看起来像随机的）

每个项目 1、2、3、4、5 似乎不是按顺序发出的。

有什么理由吗？以及如何使用 ".observeOn(Schedulers.io())" 修复它？

假设我有一个Person接口，它有一个name可观察的：

它的实现：

现在有一个问题，因为我无法获得name可观察的当前值：

我不想在我的界面中将名称公开为 BehaviorSubject，因为这样每个人都可以更改名称，这不是我想要的。

我有点理解 Observables 的逻辑，所以我知道它们不是为了存储任何值而设计的。那么什么是 RxJava 中的“具有当前值的不可变可观察对象”？

在 Android 中有一个LiveData和MutableLiveData。我的理解MutableLiveData是等价于BehaviorSubject，那么LiveData在 RxJava 中的等价物是什么？