使用BackpressureStrategy的RxJava2 Flowable.create blockingSubscribe语义
是指在使用RxJava2中的Flowable.create方法创建Flowable时，通过指定BackpressureStrategy来处理背压（Backpressure）问题，并使用blockingSubscribe方法进行订阅。
背压是指当生产者产生的数据速度快于消费者处理数据的速度时，需要一种机制来控制数据流的速度，以避免消费者被淹没或内存溢出等问题。RxJava2提供了不同的BackpressureStrategy来处理不同的背压情况。
在Flowable.create方法中，可以通过指定BackpressureStrategy来选择不同的背压策略，其中包括：
- BackpressureStrategy.ERROR：当生产者产生的数据速度快于消费者处理数据的速度时，会抛出MissingBackpressureException异常，表示背压错误。
- BackpressureStrategy.BUFFER：当生产者产生的数据速度快于消费者处理数据的速度时，会将数据缓存到一个无限大的缓冲区中，直到消费者处理完毕。
- BackpressureStrategy.DROP：当生产者产生的数据速度快于消费者处理数据的速度时，会丢弃一部分数据，以保持数据流的稳定。
- BackpressureStrategy.LATEST：当生产者产生的数据速度快于消费者处理数据的速度时，会丢弃之前产生的数据，只保留最新的数据。
使用blockingSubscribe方法可以进行阻塞式的订阅，即在订阅时会阻塞当前线程，直到数据流处理完毕或发生错误。
推荐的腾讯云相关产品：腾讯云云函数（Serverless Cloud Function），它是一种无需管理服务器即可运行代码的计算服务，可用于处理各种事件驱动的任务，包括异步消息处理、数据处理、定时任务等。腾讯云云函数支持Java语言和RxJava2库，可以方便地使用RxJava2进行异步编程和处理背压问题。
产品介绍链接地址：腾讯云云函数
相关·内容
1回答
我很难理解Flowable BackpressureStrategy是如何与blockingSubscribe方法一起工作的--或者这对我来说似乎是意想不到的，如果有人能给我解释一下，我会很感激的。我在当前主干的FlowableTests文件中测试了这段代码。@Test Flowable.create(new FlowableOnSubscribe<Integer>
浏览 0提问于2017-03-14得票数 0
回答已采纳
2回答
如何等待可流动的onComplete？
java、rx-java2
在RxJava2中，以下是我的代码：Flowable.create(e->{}, BackpressureStrategy.BUFFER我想知道如何让我的方法等待flowable onComplete，也就是说doSomething()将在flowable onCompleted之后被调用。
我找过了，但无济于事。
浏览 0提问于2019-12-20得票数 2
1回答
找不到封装方法的空lambda$getTestFlowable$0(io.reactivex.FlowableEmitter)‘
java、android、proguard、rx-java2、retrolambda
我在我的项目中使用了RxJava2和Retrolambda，一段时间以来，我感觉非常好，将回调包装为可流动的，但是有一天我遇到了这个错误，如果我试图在这个非常简单的代码片段上运行proguard：：
不要使用lambda，而是使用匿名类。私有可流动的getTestFlowable() {返回Flowable.create(新的FlowableOnSubscribe() {@覆盖公共空订阅(FlowableEmit
浏览 5提问于2017-05-12得票数 1
1回答
RxJava2从侦听器创建可流动的监听器，并在最后删除侦听器。
java、android、rx-java2
我的用例与使用RxJava2与Firebase数据库有关。disposable = Flowable.create<DataSnapshot>({ s -> s.onNext(value) })
浏览 1提问于2017-02-26得票数 1
回答已采纳
1回答
使用ForkJoinPool.commonPool()代替RxJava的计算调度程序
concurrency、rx-java2、reactive-streams
我目前正在为个人项目实现RxJava2 2的API的一个小子集。我有一个基于监听器的API，我开始编写代码来包装我的监听器并实现Flowable.create()。api.addListener(listener); }, BackpressureStrategy.BUFFER根据RxJava2 2的文
浏览 0提问于2018-03-29得票数 2
回答已采纳
2回答
可流动的单曲？
kotlin、rx-java2
尝试在rxJava2 Kotlin结合单一和流动，但什么都没有发生:不理睬什么错 emmit.onNext(id) } }
我的密码怎么了
浏览 0提问于2019-01-22得票数 0
回答已采纳
1回答
RxJava2订阅在一段时间后停止观察，但在可流动完成时继续进行。
rx-java、rx-java2
我很难理解下面的代码示例的行为； .flatMap(it -> Flowable.createe.onNext(i); }, BackpressureStrategy.BUFFER发射和观察是平行的。但在此之后，可流继续发出项(这些项显然在某个位置排队)，但在发出所有1000
浏览 1提问于2017-07-05得票数 4
回答已采纳
1回答
Rx 2:如何包装回调？
java、android、callback、rx-java2
我有这段代码来包装Rx 1中的回调，它编译得很好，但是现在我已经切换到RX 2了，它不是compile...what在Rx 2中是等价的？} }更新：
这是我想出来的，
浏览 4提问于2017-04-05得票数 11
1回答
Rx Java2处理异步超时异常
java、rx-java2
我在做一些并行的io手术。下面的代码是我的架构的测试代码。我想给异步io操作超时，但是流在一些步骤之后就会停止。当我放弃超时功能代码时，效果很好。 Flowable.create.subscribeOn(flatmapSubscribeOn) .subs
浏览 1提问于2018-12-17得票数 0
回答已采纳
1回答
自定义RxJava Observable在订阅时不发出任何内容
observable、rx-kotlin
reader.readLine() } return stringBuilder.toString() } catch (e: IOException) { }} 但它并没有像我预期的那样工作print("Error: $it") .doOnComplete {
print("completed
浏览 11提问于2018-12-24得票数 0
1回答
rxjava2 -如何阻止，直到回叫完成，然后进行改造调用？
rx-java、rx-java2
我很难使用BlockingObservable。我有一种情况，我需要回叫已完成，然后在它完成之后，我可以做一个修改API调用来获取数据。subscribeOn(AndroidSchedulers.mainThread())rxjava2因此，明确地说，当我调用paymentStrategy.init()时，我需要可以观察到的阻塞，直到用户onComplete或onNext被调用。
浏览 9提问于2017-07-29得票数 0
1回答
如何用新的RxJava2替换返回的observable
android、rx-java、observable、rx-java2
startStreamRead() { return mFlowableStream;然后在绑定到服务之后，我为它提供一个类似这样的回调连接public void accept(byte[] data) throws Exception {
});我想要做的是以某种方式将以前返回的mFlowableStream替换为我从服务中获得的新观察值。有什么可能<
浏览 0提问于2017-11-15得票数 0
1回答
是什么取代了rxjava2中的AsyncEmitter？
rx-java
我通过指定一个BackpressureMode，使用一个AsyncEmitter连接到一个现有的提要。然而，我没有在rxjava2中找到一种类似的方法来做到这一点。是什么取代了rxjava2中的AsyncEmitter？我看到有许多发射器实现，但没有一个接受BackPressureStrategy。我试过了 new FlowableCreate(source, BackpressureStrategy.L
浏览 27提问于2017-01-30得票数 2
1回答
RxJava2可观测背压
java、rx-java、rx-java2、backpressure
最近我意识到我不明白RxJava2背压是如何工作的。doOnNext(i -> { System.out.println("Processed:" + i); .blockingSubscribe我希望e.onNext(i);会将项目放入ObservableObserveOn的缓冲区，并且在它的大小大于static final int BUFFER_SIZE = Math.max(16,I
浏览 0提问于2017-06-21得票数 6
回答已采纳
1回答
FlowableEmitter无法发出项目
rx-java2
我有一个程序，轮询2个目录- dir1和dir2和发射的文件，只要文件到达这2个目录中的任何一个。但是，当文件同时到达这两个目录时，其中任何一个目录中的某些文件都无法发出。代码如下： Flowable.create((FlowableEmitter<Path> em) -> pollDirectory(Arrays.asList(dir1,dir2), em),BackpressureStrategy.BUFFER).subscribe(); pollDirectory方法的</
浏览 36提问于2021-07-27得票数 0
1回答
如何在Apache方法中使用NiFi？
rx-java、apache-nifi
我会从FlowFile中读取字符并使用移位缓冲区发布它们；例如.emitter.onComplete(); });
我也试过使用Flowable<Tuple<Long, Integer>> chars = try {
intemit
浏览 0提问于2017-04-06得票数 1
回答已采纳
1回答
添加RxJava2后，应用程序在SocketTimeout上崩溃
android、retrofit2、rx-java2
在我的应用程序之前，我按顺序进行了两个API调用：isCategoryPromoted和loadCategoryServices。我已经看过StackOverflow了，似乎几乎所有的问题都表明只是增加了会话超时时间。我相信这是一个黑客，应该有一个完全合理的方法来解决这个问题。CategoriesLocalDatsSource.kt
ov
浏览 1提问于2017-11-13得票数 0
回答已采纳
1回答
如何同步和异步运行同一函数
rx-java2、android-room、android-viewmodel、android-mvvm
我在安卓应用程序中使用了Room和RxJava2。我有一个场景，在这个场景中，我希望使用相同的函数在某个点运行synchronously，并在其他地方运行asynchronously。任何有效的方法都将受到赞赏。
谢谢
浏览 2提问于2019-06-24得票数 0
1回答
如何向ConnectableFlowable发送取消信号？
java、rx-java、flowable、disposable
我使用一次性的emit来发布和订阅项目。但是当我尝试使用ConnectableFlowable时，我无法向发射器发送取消信号。我如何理解flowable是在Flowable.create方法中处理的？ 您可以通过注释和取消注释“publish().autoConnect()”代码片段来查看该场景。Disposable disposable = Flowable.create(emitter -> {
AtomicBoolean isRunning = new Ato
浏览 15提问于2020-12-07得票数 4
回答已采纳
1回答
仅当表大小更改时通知(RX)数据观察者
android、objectbox
我使用响应式观察者(RxJava2)来监听数据库的变化。> { return RxQuery.observable(query).toFlowable(BackpressureStrategy.BUFFER
浏览 18提问于2019-01-25得票数 0
腾讯云开发者
