RxJava3。为什么没有调用FlowableSubscriber onNext？
RxJava3 中 FlowableSubscriber 的 onNext 未被调用的原因及解决方法
基础概念
RxJava 是一个用于处理异步数据流的库，它提供了丰富的操作符来处理数据流。Flowable 是 RxJava3 中用于处理背压（backpressure）的接口，适用于数据量较大或生产速度大于消费速度的场景。
相关优势
- 背压处理：Flowable 能够处理生产者和消费者之间的速度不匹配问题，避免数据丢失或内存溢出。
- 操作符丰富：提供了大量的操作符来处理数据流，如 map、filter、flatMap 等。
- 异步处理：支持异步操作，提高应用的响应性和性能。
类型与应用场景
- FlowableSubscriber：用于订阅 Flowable 的接口，包含 onNext、onError 和 onComplete 方法。
- 应用场景：适用于需要处理大量数据流、实时数据处理、网络请求等场景。
问题原因
FlowableSubscriber 的 onNext 方法未被调用可能有以下几种原因：
- 数据未生成：生产者没有生成数据，或者数据生成逻辑有误。
- 订阅未成功：订阅逻辑有误，导致订阅未成功。
- 背压问题：生产者生成数据的速度远大于消费者处理数据的速度，导致数据丢失或未被处理。
- 线程问题：数据处理逻辑在不同的线程中执行，可能导致 onNext 方法未被调用。
解决方法
以下是一些常见的解决方法：
- 检查数据生成逻辑： 确保生产者能够正确生成数据，并且数据生成逻辑没有问题。
- 检查数据生成逻辑： 确保生产者能够正确生成数据，并且数据生成逻辑没有问题。
- 确保订阅成功： 确保订阅逻辑正确，订阅成功后才会调用 onNext 方法。
- 确保订阅成功： 确保订阅逻辑正确，订阅成功后才会调用 onNext 方法。
- 处理背压问题：
使用合适的背压策略，如
BackpressureStrategy.BUFFER、BackpressureStrategy.DROP等。 - 处理背压问题：
使用合适的背压策略，如
BackpressureStrategy.BUFFER、BackpressureStrategy.DROP等。 - 检查线程问题：
确保数据处理逻辑在正确的线程中执行，可以使用
subscribeOn和observeOn方法来指定线程。 - 检查线程问题：
确保数据处理逻辑在正确的线程中执行，可以使用
subscribeOn和observeOn方法来指定线程。
参考链接
通过以上方法，可以解决 RxJava3 中 FlowableSubscriber 的 onNext 方法未被调用的问题。
相关·内容
我有一个git主分支和功能分支。主分支使用rxjava3，但功能分支使用rxjava2。然而，当我从主分支切换到功能分支时，Android Studio无法识别rxjava2导入。有趣的是，即使有所有标记的错误(无法解析符号)，它也可以很好地构建。 ? 如果我请求Android Studio导入，它会尝试导入rxjava3 (import io.reactivex.rxjava3.*)。 如何让Android Studio在不尝试导入rxjava3的情况下识别java2导入？我的gradle构建引用了rxjava2，而不是rxjava3，所以我很困惑为什么它构建得很好，但我说有一堆未
浏览 111提问于2021-10-09得票数 1
回答已采纳
我正在使用RxJava3和改造，但是我无法为RxJava3获得一个rxjava3改造适配器。
浏览 4提问于2020-03-09得票数 28
回答已采纳
我实现了一个FlowableOperator，如RxJava2 wiki ()中所描述的，除了在onNext()操作中执行一些测试之外，类似于这样：
public final class MyOperator implements FlowableOperator<Integer, Integer> {
...
static final class Op implements FlowableSubscriber<Integer>, Subscription {
@Override
public void onNext(Integer v) {
浏览 0提问于2018-02-22得票数 0
回答已采纳
4回答
房间支持RxJava 3
、、、
我在项目中使用带有文件室的RxJava3，但收到以下错误
错误:不确定如何将游标转换为此方法的返回类型(io.reactivex.rxjava3.core.Flowable>)
下面是我收到错误的DAO接口方法
@Query("SELECT * FROM wishlist_table")
Flowable<List<WishListMovie>> getWishList();
我想这可能是因为我在成绩文件中使用了下面的依赖项：
implementation "androidx.room:room-rxjava2:$room_version&
浏览 5提问于2020-06-14得票数 6
为什么/如何在没有Connect或RefCount调用的情况下使用.Publish()？是干什么的呢？示例代码：
var source = new Subject<int>();
var pairs = source.Publish(_source => _source
.Skip(1)
.Zip(_source, (newer, older) => (older, newer))
);
pairs.Subscribe(p => Console.WriteLine(p));
source.OnNext(1);
source.OnNext(2)
浏览 1提问于2018-12-13得票数 1
1回答
下面是测试代码
final Flowable<Integer> f1 = Flowable.fromPublisher(s -> {
s.onNext(Integer.valueOf(1));
s.onComplete();
});
final Flowable<Integer> f2 = Flowable.fromPublisher(s -> {
s.onNext(Integer.valueOf(2));
s.onComplete();
});
F
浏览 0提问于2016-12-10得票数 0
1回答
我是RxAndroid的新手，这就是为什么提到Rx的所有运算符的原因。我发现observable.takeFirst()显示红线，因为在RxJava3中找不到它。谁能告诉我takefirst()方法的替代方法是什么？
浏览 13提问于2020-09-11得票数 2
在我的Swift UIViewController中，我尝试订阅一个变量类型的类成员，通过flatMapLatest调用运行它，然后在所有订阅者上执行flatMapLatest observable中的onCompleted()调用。然而，当onNext()被调用时，onCompleted()却从来没有被调用过，我不知道为什么。
我的类成员定义为：
private let privateVar = Variable<String>("")
在我的viewDidLoad()方法中，我设置了观察值：
let localVar = self.privateVar.asOb
浏览 0提问于2018-08-10得票数 1
RxAndroidBle是一个很好的软件，它减少了Ble项目的开发时间，并且显着地提高了稳定性和可读性。
我只想问一问，rxjava3的RxAndroidBle分支的状况如何？它说，这是一个试验分支，并在这个时候陈腐。因此，不建议将rxjava3分支用于生产应用程序吗？
我不知道将来是否有计划专注于rxjava3，因为RxJava页面上写着：
2.x版本处于维护模式，只会通过补丁支持，直到2021年2月28日。
来自t4rj4n的问候
浏览 0提问于2020-05-14得票数 2
1回答
在阅读了大量关于堆栈溢出的帖子之后，我仍然对Rx中可观察的和主题的区别感到非常困惑。
基本上，大多数人都指出，两者之间的主要区别是“ Subject类继承了可观察的和观察者的，也就是说它既是观察者又是可观察的”，可以通过下面的代码片段来演示：
var subject = new Rx.Subject();
var subscription = subject.subscribe(
function (x) { console.log('onNext: ' + x); },
function (e) { console.log('onError: ' + e.m
浏览 4提问于2019-12-11得票数 1
回答已采纳
为什么在subject.onNext(o)中调用doOnSubscribe没有任何效果，但是调用subject.onComplete()会导致流终止！？
final PublishSubject<Integer> subject = PublishSubject.create();
final Observable<Integer> observable = subject.doOnSubscribe(new Consumer<Disposable>() {
@Override
public void accept(
浏览 3提问于2017-04-06得票数 1
回答已采纳
下面的代码 为什么onnext只调用了一次？当我删除subscribeOn时，每个数字都会调用它。当I subscribeOn io线程只调用一次时(对于8658) 有人能给我解释一下吗？ val subject = BehaviorSubject.create<Int>()
subject.onNext(2121)
subject.distinctUntilChanged().doOnNext {
Log.d("AHMET VEFA SARUHAN", it.toString())
}.subscribeOn(Sc
浏览 9提问于2020-12-11得票数 0
当用户在TextView中输入文本时，我试图进行一些搜索：
RxTextView.textChangeEvents(mTextView)
.doOnNext(new Action1<CharSequence>() {
@Override
public void call(CharSequence charSequence) {
System.out.println("OnNext: " + charSequence);
}
})
.observeOn(mSchedulerProvider.computation())
浏览 3提问于2015-11-24得票数 0
回答已采纳
下面的
var subject = new Subject<int>();
var otherSubject = new Subject<Unit>();
subject
.SkipUntil(otherSubject)
.Subscribe(Console.WriteLine, () => Console.WriteLine("Completed"));
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
otherSubject.OnNext(Unit.Default);
s
浏览 0提问于2017-03-31得票数 1
回答已采纳
如果在main()方法中，我执行以下操作
Flux.just(1,2)
.log()
.subscribe();
我在控制台上看到这个：
[ INFO] (main) | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription)
[ INFO] (main) | request(unbounded)
[ INFO] (main) | onNext(1)
[ INFO] (main) | onNext(2)
[ INFO] (main) | onComplete()
如果我没有使用just()，而是使用了int
浏览 8提问于2020-05-13得票数 2
回答已采纳
Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
subscriber.onStart();
subscriber.onNext(1);
subscriber.onCompleted();
}
}).delaySubscription(5, TimeUnit.SECONDS).subscribe(n
浏览 2提问于2017-01-21得票数 1
我想使用rxjava 1测试merge操作符，使用：
Observable<Object> o1 = Observable.create((t) -> {
t.onNext("1");
try {
Thread.sleep(5000);
} catch (Exception e) {
e.printStackTrace();
}
t.onNext("2");
t.onCompleted();
浏览 4提问于2017-01-19得票数 0
回答已采纳
我正在尝试在C#中使用.NET的RX库。谁能给我解释一下为什么'observer.OnCompleted()‘方法在下面的代码中什么也不做：
var observableStream = Observable.Create<CustomMessage>(
(observer) =>
{
CustomMessage cm = new CustomMessage();
CustomMessage.Subscribe(observer.OnNext);
浏览 5提问于2012-11-26得票数 1
回答已采纳
1回答
我正在尝试将我的firebase项目转换为使用rx swift，但是当onCompleted触发器在subscription中时，.do(onNext:{})会被丢弃。
我对observable代码做了一个firebase函数：
func fetchFireBaseData() -> Observable<[Recipe]> {
return Observable.create({ [weak self] observer -> Disposable in
guard let self = self else { return Disposabl
浏览 2提问于2020-10-12得票数 1
回答已采纳
2回答
我正在读一本关于RxJava的书，下面是其中的一段摘录： Observable.create(s -> {
... async subscription and data emission ...
})
.doOnNext(i -> System.out.println(Thread.currentThread()))
.filter(i -> i % 2 == 0)
.map(i -> "Value " + i + " processed on " + Thread.currentThread())
.subscribe(s -
浏览 20提问于2019-01-10得票数 1
回答已采纳
我需要获取一个订阅对象，才能获得取消订阅侦听器的机会。为此，我想给函数提供一个FlowableSubscriber。 代码： FlowableSubscriber fs = new FlowableSubscriber() {
@Override
public void onSubscribe(@NonNull Subscription s) {
System.out.println("Flowable onSubs");
}
@Override
public void
浏览 120提问于2020-11-10得票数 1
回答已采纳
下面是我的代码： package com.example.myapplication;
import io.reactivex.Observable;
public class SampleRx {
Observable<String> getBoth() {
return Observable.merge(getSeq1(), getSeq2());
}
Observable<String> getSeq1() {
return Observable.create(emitter -> {
浏览 27提问于2019-09-02得票数 0
回答已采纳
我有一个网络调用，它返回一个项目列表。我希望继续重复该网络调用，直到项目数与初始计数不同，或者5秒已经过去。我试过这个：
timer = 0;
while (initialCount== afterCount && timer != 5) {
try {
Thread.sleep(1000);
timer++;
Observable<List<Item>> itemObservable = getItemsFromNetwork();
itemObservable.subscribe(new Observer<L
浏览 4提问于2015-03-15得票数 1
1回答
在filter方法之后，subscribe方法似乎给出了一个nullpointerexception。当我删除filter方法时，一切都运行正常。为什么？
Flux<String> flux = new Flux<String>() {
@Override
public void subscribe(CoreSubscriber<? super String> coreSubscriber) {
coreSubscriber.onNext("Test 7&
浏览 0提问于2019-11-24得票数 0
我很难弄清楚SequenceEqual如何在ReactiveExtension2.2.5中工作
下面是代码
var subject1 = new Subject<int>();
var subject2 = new Subject<int>();
IObservable<bool> sequenceEqual = subject1.SequenceEqual(subject2);
subject1.OnNext(5);
subject1.OnNext(3);
浏览 0提问于2015-04-29得票数 1
回答已采纳
1回答
为什么下面的代码不会产生任何结果？这个Subject中出了什么问题？我希望在我订阅的那一刻会得到一个结果，也就是我声明的长度。 val subject = PublishSubject.create<String>()//PublishRelay.accept() does not work as well
subject.onNext("Alpha"
subject.onNext("Beta")
subject.onNext("Gamma")
subject.onComplete()
subject.map{ it.length
浏览 15提问于2019-12-19得票数 1
回答已采纳
1回答
Flux.just("a", "b", "c")
.log(null, Level.INFO,true) // line 18
.flatMap(value -> Mono.just(value.toUpperCase())
.publishOn(Schedulers.elastic()), 2)
.log(null, Level.INFO,true) // line 21
.subscribe();
其中一些产出：
浏览 1提问于2018-02-17得票数 2
回答已采纳
下面是简短的代码片段：
val subject = BehaviorSubject.createDefault(emptyList<Int>())
subject.onNext(Arrays.asList(1, 2, 3))
subject.flatMapIterable { list: List<Int> -> list }
.subscribeBy(
onNext = { l("on next", it) },
onC
浏览 3提问于2017-10-05得票数 0
回答已采纳
3回答
等待观察完成
、、、、
我有一种方法，用可观测的方法异步地做一些工作。我想知道使这个方法异步的最佳方法是什么，这样我就能够等待它并在可观察完成之后做一些工作。
我的第一次尝试是使用等待观察。
public async Task DoWorkAsync()
{
var observable = Observable.Create<int>(o =>
{
Task.Run(() =>
{
Thread.Sleep(1000);
Console.WriteLine("OnNext");
浏览 3提问于2017-02-14得票数 3
阅读一篇关于可观察的文章，发现代码在问题的末尾。
我无法理解以下几行->
return this._subscribe({
onNext: onNext,
onError: onError || (() => {}),
onCompleted: onCompleted || (() => {})
});
1)我以前从未见过这种语法，这到底是什么？
使用它的类型表示它是一个对象，但在我看来，它就像函数中的一个对象，这是没有意义的。
2)既然我不懂代码，到处玩就
浏览 2提问于2020-05-09得票数 0
回答已采纳
根据MSDN文档，以下代码应将'5‘输出到控制台窗口。相反，不会显示任何内容。
static void Main(string[] args)
{
var o = new ReplaySubject<int>();
o.OnNext(0);
o.OnNext(1);
o.OnNext(2);
o.OnNext(3);
o.OnNext(4);
o.OnNext(5);
o.TakeLast(1).Subscribe(Console.WriteLine);
Console.Writ
浏览 0提问于2011-07-20得票数 1
回答已采纳
我想要一个不做任何事情的Observable，当订阅时，会调用观察者的onNext回调函数。我想我在Observable.empty()中发现了这一点，但是下面的代码并没有导致观察者的onNext回调被调用：
Observable<Void> emptyObservable = Observable.empty();
emptyObservable.subscribe(passedinObserver);
定义了passedInObserver：
Observer<Void> passedinObserver = new Observer<Void> {
浏览 0提问于2015-03-13得票数 0
3回答
我有一个可以观察到的东西，它不需要发出一个值。此外，我有一个对象的列表，我希望可以观察到的工作。因此，对于这个列表中的所有元素: doSomething()
Observable.from(uris)
.flatMap(new Func1<Uri, Observable<Void>>() {
@Override
public Observable<Void> call(Uri uri) {
return createDoSomethingObservable(ur
浏览 1提问于2015-07-23得票数 13
回答已采纳
Kotlin 1.3.72，RxJava2
我有以下代码，我试图避免使用！运算符，但不确定为什么它认为值为null，我需要使用安全调用操作符。
以后我要用这个！这是很糟糕的做法。为什么这是空的，因为我已经声明了任何可以为空的类型？
class SharedViewModel : ViewModel() {
private val compositeDisposable = CompositeDisposable()
private val imageSubject = BehaviorSubject.create<MutableList<Photo>>(
浏览 9提问于2020-08-10得票数 7
回答已采纳
我正在学习RxJS。我对下面的代码片段有疑问。
var arr = [1,2,3,4,5];
var arraysource = Observable.from(arr);
arr.push(6);
var subscription = arraysource.subscribe(
x => console.log('onNext: %s', x),
e => console.log('onError: %s', e),
() => console.log('onCompleted'));
ar
浏览 2提问于2016-06-09得票数 2
回答已采纳
我想知道为什么cancel信号的处理方式与其他信号不同。有人知道这是否是故意的吗？ Scheduler scheduler = Schedulers.elastic();
Disposable disposable =
Flux.interval(Duration.ofSeconds(1))
.publishOn(scheduler)
.doFinally(signalType -> {
if (signalType == SignalType.CANCEL) {
// caller thread
浏览 10提问于2019-01-28得票数 2
回答已采纳
我是RxJava的新手，我无法意识到--为什么我的“压缩”可观察功能在使用两个PublishSubject时不会发出项呢？(据我所知，ZIP运营商应该将两个流“合并”为一个)
val currentSubject = PublishSubject.create<Int>()
val maxSubject = PublishSubject.create<Int>()
currentSubject.onNext(1)
maxSubject.onNext(2)
currentSubject.onNext(1)
maxSubject.onNext(2)
Log.d(
浏览 2提问于2019-06-25得票数 0
回答已采纳
我有个风景
subject1: PublishSubject和subject2:BehaviorSubject.
首先，我为subject1发出单个项，然后为subject2发出项，但在此之后，我还想向subject1发出不同的项。
fun emittingItems() {
subject1.onNext(functionA1)
subject2.onNext(functionB)
if (something) subject1.onNext(functionA2)
}
所发生的是，我收到了一个项目的顺序：functionA1，functionA2，functionB
浏览 5提问于2018-05-31得票数 0
回答已采纳
3回答
我是反应式开发的新手，在组合多个主题的方式上遇到了困难，这种方式将授予初始值以及随着时间的推移在主题上发生的变化。
下面的代码是我所期望的：
first - OnNext(a) <-- initial on subscribe
first - OnNext(c) <-- initial on subscribe
first - OnNext(b) <-- result of OnNext
second - OnNext(b) <-- initial on subscribe
second - OnNext(c) <-- initial on subscrib
浏览 10提问于2017-08-23得票数 3
回答已采纳
我是RxSwift和反应式编程的新手。下面是我的代码：
let disposeBag = DisposeBag()
let dartScore = PublishSubject<Int>()
dartScore.asObserver()
.scan(501, accumulator: {
intermediate, newValue in
var result = intermediate - newValue
return result >= 0 ? result : intermediate
})
.do
浏览 14提问于2016-09-03得票数 0
使用subscribeOn 排除运行以下代码
PublishSubject<Void> repeatSubject = PublishSubject.create();
Observable.
<Integer>create(subscriber -> {
System.out.println("1");
subscriber.onNext(1);
subscriber.onCompleted();
}).
//subscribeOn(Schedulers.computati
浏览 5提问于2017-05-03得票数 0
可观察对象必须连续地(不是并行地)向观察员发出通知。它们可能从不同的线程发出这些通知，但必须在通知之间的关系之前正式发生。
当它说“发生之前”时，是否意味着上一次onNext通知(例如，更改Observer.onNext()方法中的共享状态)所产生的所有影响在下一个onNext通知中将完全可见，就像在Java 中的保证之前发生的一样
在阅读了SerializedObserver的源代码(RxJava版本3.0.11)后，我发现同步代码块中没有调用downstream.onNext(t)方法。所以，我可以说答案不是吗？
浏览 5提问于2021-04-22得票数 0
回答已采纳
我是一个新的响应式程序员。我想在defer中调用观察者的onNext方法，有没有好的方法呢？
举个例子-
Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super String> observer) {
try {
if (!observer.isUnsubscribed()) {
// While using create, can
浏览 3提问于2017-01-18得票数 1
试图了解Subject<T>、ReplaySubject<T>和其他工作方式。下面是一个例子：
(主题是可观察的和观察者)
public IObservable<int> CreateObservable()
{
Subject<int> subj = new Subject<int>(); // case 1
ReplaySubject<int> subj = new ReplaySubject<int>(); // case 2
Rando
浏览 1提问于2012-08-20得票数 3
回答已采纳
我试图使用testNG运行套件，但是我发现了一个错误:我在这个站点和网站上搜索了一些关于这个问题的信息，但是没有成功。
[RemoteTestNG] detected TestNG version 7.4.0
java.lang.NoClassDefFoundError: io/reactivex/rxjava3/core/Observer
at resources.ExtentReporterNG.getReportObject(ExtentReporterNG.java:14)
at ShufersalMiron.Sprod.Listeners.<init>(L
浏览 6提问于2022-02-06得票数 -1
如果可观测数据是由有限的数据构成的，它们就自然地完成了。
import {Observable, Subject} from "rx";
let stream0$ = Observable.of("1", "2", "3");
let stream1$ = stream0$.map(x => x);
stream1$.subscribe(
(val) => { console.log("onNext", val) },
(err) => { console.log("onErr
浏览 6提问于2016-01-18得票数 2
回答已采纳
为什么subscribe从不在这里打印任何东西？只是出于好奇。无论如何，这是一种糟糕的做法:我通常会使用observeOn。然而，我不明白为什么subscribe从来没有到达.
val subject: PublishSubject<Int> = PublishSubject.create()
val countDownLatch = CountDownLatch(1)
subject
.map { it + 1 }
.subscribeOn(Schedulers.computation())
.subscribe {
println(Th
浏览 2提问于2020-08-09得票数 1
回答已采纳
我正在处理Flux和Mono，并在多线程环境中使用它们，并使用提供工作线程的调度器。
有许多选项可以使用弹性、并行和newElastic启动调度。
下面是我使用的代码：
System.out.println("------ elastic --------- ");
Flux.range(1, 10)
.map(i -> i / 2)
.publishOn(Schedulers.elastic()).log()
.blockLast();
System.out.println("------
浏览 1提问于2018-01-04得票数 5
回答已采纳
1回答
我正在使用RxUI 8的最新预发行版，但我认为这会发生在以前的版本中。
我在我的WPF应用程序中定义了这个ReactiveCommand：
GetWebsiteCommand = ReactiveCommand.CreateFromTask(DownloadString);
private async Task<string> DownloadString()
{
using (var client = new WebClient())
{
return await client.DownloadStringTaskAsync("http:
浏览 1提问于2017-10-04得票数 4
回答已采纳
3回答
我试图理解下面的代码，尝试使用反应性扩展
IObservable<string> textChangedObservable =
Observable.FromEventPattern<TextChangedEventArgs>(searchScrip, "TextChanged")
.Select(evt => ((TextBox) sender).Text);
textChangedObservable.Subscribe(OnNext, OnCompleted);
private
浏览 5提问于2015-09-04得票数 6
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
