我想创建一个可以将对象推送到其中的buffering Multi。Multi用作REST端点的结果。private PublishProcessor<String> deviceStatusProcessor = PublishProcessor.create(); }
return Multi.createFro
浏览 39提问于2020-04-25得票数 0
回答已采纳
如果我按照反应性规范实现自己的Publisher<T>：现在，我想将它与rxJava2可流连接起来，两者之间的区别是：和
Flowable.fromPublisher(
浏览 1提问于2018-02-28得票数 0
回答已采纳
1回答
我很难找到任何RxJava2的例子，把两个流量计拉链成一个。<Integer> source = Flowable.fromArray(ints); }
编辑:我试图从源和<
浏览 0提问于2017-10-09得票数 5
回答已采纳
2回答
我正在尝试将“第三方”publisher封装到RxJava中。在订阅期间，我正在准备publisher，这可能会导致异常(当publisher不可用时)。2)将其推入观察者的OnError方法？我正在考虑推送OnError，那么如何实现订阅呢？如果OnError应该是subscription中的最后一个事件，我应该将其视为subscrip
浏览 2提问于2017-01-26得票数 0
对于下面的代码，我得到了以下错误：@DaoFlowable<List<Friends>> getAllFriends();
Observable<List<Friends>> getAllFriendsExample
浏览 0提问于2018-06-16得票数 0
回答已采纳
根据introtorx.com的说法，Observable.subscribe(Obeserver)返回Disposable。我在Android Studio中使用compile 'io.reactivex.rxjava2:rxandroid:2.0.1'。
浏览 5提问于2017-06-16得票数 6
我目前正在使用以下代码将Akka Source (例如使用Akka的FileIO读取文件时收到的)转换为可流动的RxJava2： Flow<ByteString, ByteString, NotUsed> compType) {
final Publisher(uncompressed
浏览 9提问于2017-08-24得票数 1
回答已采纳
然而，尝试在Kotlin中将重试添加到RxJava2 retrofit2调用中，无法找出正确的语法。我做了一些研究，发现我需要使用类似于模式的retryWhen (用于重试3次)：api
.subscribeOn(Schedulers.io: Flowable<(???..???)>!> zipWith(p0: Publisher<out (???..???)>!
浏览 1提问于2018-09-04得票数 0
我正在学习如何在RxJava2中进行数据轮询。 compositeDisposable.add(getMyTask()
.repeatWhen(new Function<Flowable<Object>,Publisher<?在我的Observable中，这是不产生的，这是有原因的，因为我正在测试逻辑，只<e
浏览 0提问于2018-03-24得票数 1
回答已采纳
我试图为我的本地db使用空间，我的问题是，我的第一个应用程序屏幕显示了保存在房间库中的设备列表，但是房间不提供任何回调，比如成功或失败，那么你们是如何处理的呢?我还可以在从房间加载数据时显示加载程序，因为在ui线程上运行之前，我不能从线程中运行ui，而这是房间不推荐的。
浏览 1提问于2018-12-27得票数 1
回答已采纳
在Rxjava1中，我们可以订阅一个可观察对象，如下所示：Observable<T>.subscribe(asyncSubject不知道如何在Rxjava2 Flowable中实现同样的功能？我在想像这样的事情.doOnComplete { t -> asyncSubject.OnComplete
浏览 6提问于2019-07-23得票数 0
回答已采纳
在我的设置中，我希望通过SSE通道(服务器发送的事件)转发某些状态更改。状态更改是通过调用REST端点启动的。因此，我需要将传入的状态更改转发到SSE流。在Quarkus中完成此操作的最佳/最简单方法是什么。 我能想到的一种解决方案是使用EventBus (https://quarkus.io/guides/reactive-messaging)。状
浏览 24提问于2020-04-23得票数 2
回答已采纳
1回答
RxJava2可观测背压
、、、
最近我意识到我不明白RxJava2背压是如何工作的。我希望e.onNext(i);会将项目放入ObservableObserveOn的缓冲区，并且在它的大小大于static final int BUFFER_SIZE = Math.max(16,Integer.getInteger("rx2.buffer-size",128).intValue());之后
它应该抛出MissingBackpressureException，这是不会发生的。
浏览 0提问于2017-06-21得票数 6
回答已采纳
1回答
一个简单的例子：import org.reactivestreams.Publisher; Thread.sleep(time); }
所以问题是，我想异步调用方法m1()和m2上，他们有一些例子，如何在新线程中运行：
import io.reactivex.schedul
浏览 0提问于2018-03-10得票数 1
我在我的项目中使用了RxJava2和Retrolambda，一段时间以来，我感觉非常好，将回调包装为可流动的，但是有一天我遇到了这个错误，如果我试图在这个非常简单的代码片段上运行proguard：private Flowable<Object> getTestFlowable() {
return Flowable.create(emitter -> new Thread(newcom.yandex.testfield
浏览 5提问于2017-05-12得票数 1
1回答
List<Action> handle() {} executeActionsAsync(actions) { action.forEach(functionexecuteAsync(actions);}
我想在RXJava2中做同样的模型。任何帮助都是非常感谢的。
浏览 0提问于2018-09-19得票数 8
我已经使用RxJava2很长时间了。最近，我正在研究RxJava2的源代码，试图理解它，编写我自己的运算符实现。我发现有很多线程安全的代码。但我认为有些是多余的，这增加了学习的难度。根据 2.7，“订阅者必须确保其订阅上的所有调用都发生在同一线程中，或者提供各自的外部同步。”调用者不是有责任确保它的调用是线程安全的吗？但是我发现"io.reactivex.internal
浏览 0提问于2018-07-27得票数 1
3回答
我有一个可以观察到的东西，它可以从很多来源获得：资源和项目之间的关系是多对多的，在不同的来源中，项目可能重复自己。项是一个应该上载到服务器并且服务器不接受重复项的实体。为了实现这一点，我合并源并根据它们的ids区分它们的项，然后将唯一的项上传到服务器。我如何推迟上传“失败”项目的尝试，并继续进行其他项目，而这个项目正在等待它
浏览 3提问于2017-09-11得票数 2
回答已采纳
1回答
微控制器中不耗尽内存的流大响应
、、、、
我们使用Micronaut和Mongo通过一些控制器公开数据。由于响应实体的大小正在增长，我们的应用程序有时会耗尽内存。{ "field": "value" }我们使用这个控制器工作，其中dataStore返回一个Publisher<Example> Flowable<);这样做很好，在将大量示例流到客户端之前，不必将其完全加载到内存中</em
浏览 0提问于2021-03-06得票数 2
回答已采纳
如何让RxJava在NiFi内部工作？？或者我如何让NiFi和RxJava表现得很好？它们似乎是彼此完美的补充。 }, BackpressureStrategy.BUFFER);
浏览 0提问于2017-04-06得票数 1
回答已采纳
云服务器
ICP备案
对象存储
云直播
腾讯会议
腾讯云开发者
