自从我集成 RxJava2 以来，我在返回Observable 的所有改造调用中收到 401 未经身份验证的错误，我正在使用基本身份验证，并且我知道错误是由于它造成的，但为什么它在调试时有效，但在发布时无效。

在我看来，retrofit2 的 rxjava 适配器的配置有问题

堆栈跟踪：

com.jakewharton.retrofit2.adapter.rxjava2.HttpException: HTTP 401 Unauthorized
01-22 19:24:14.872 11502-11502/? W/System.err:     at com.jakewharton.retrofit2.adapter.rxjava2.BodyObservable$BodyObserver.onNext(BodyObservable.java:54)
01-22 19:24:14.872 11502-11502/? W/System.err:     at com.jakewharton.retrofit2.adapter.rxjava2.BodyObservable$BodyObserver.onNext(BodyObservable.java:37)
01-22 19:24:14.872 11502-11502/? W/System.err:     at com.jakewharton.retrofit2.adapter.rxjava2.CallObservable.subscribeActual(CallObservable.java:43)
01-22 19:24:14.872 11502-11502/? W/System.err:     at io.reactivex.Observable.subscribe(Observable.java:10514)
01-22 19:24:14.872 11502-11502/? W/System.err:     at com.jakewharton.retrofit2.adapter.rxjava2.BodyObservable.subscribeActual(BodyObservable.java:34)
01-22 19:24:14.872 11502-11502/? W/System.err:     at io.reactivex.Observable.subscribe(Observable.java:10514)
01-22 19:24:14.872 11502-11502/? W/System.err:     at io.reactivex.internal.operators.observable.ObservableDoOnLifecycle.subscribeActual(ObservableDoOnLifecycle.java:33)
01-22 19:24:14.872 11502-11502/? W/System.err:     at io.reactivex.Observable.subscribe(Observable.java:10514)
01-22 19:24:14.872 11502-11502/? W/System.err:     at io.reactivex.internal.operators.observable.ObservableDoOnEach.subscribeActual(ObservableDoOnEach.java:42)
01-22 19:24:14.872 11502-11502/? W/System.err:     at io.reactivex.Observable.subscribe(Observable.java:10514)
01-22 19:24:14.872 11502-11502/? …

我在(Rx)Java中有这个代码:

Observable.fromArray(1, 2, 3)
    .flatMap(this::intToBooleanObservable, Pair::new)
    .....

我希望相应的Kotlin代码看起来像:

Observable.fromArray(1, 2, 3)
    .flatMap(::intToBooleanObservable, ::Pair)
    .....

但是编译器无法推断出Pair的泛型类型,所以我现在能做的最好的事情是:

.flatMap(::intToBooleanObservable, { a, b -> a to b })

这并不像我希望的那样简洁.

有没有办法实现这个不声明变量a和b？

我有一个Observable，我想定期重复，但仅限以下情况：

apiInterface.getData() // returns Observable<Data>
... // processing is happening here
.toList()
.repeatWhen(completed -> {
    if (autoReload){
        // Repeat every 3 seconds
        return completed.delay(3, TimeUnit.SECONDS);
    } else {
        return ??? // What do I have to return that it does not repeat?
    }
})
.subscribe(list -> callbackInterface.success(list));

我的问题是：我必须在else语句中返回什么才能不重复Observable（只需执行一次链）？

我做了一个RxJava2实验,包括以下步骤:

1. 从ListA的对象创建一个Observable .
2. 使用flatMapSingle()将Observable发出的每个项映射到SingleSource.在flatMapSingle()内部,对Observable发出的每个项执行异步操作.
3. 使用Observable.toList()收集ListB中的所有项

结果:数组listB有其不同的顺序项目利斯塔由于异步操作.

因此,似乎flatMapSingle()的工作方式类似于flatMap(),因为它使用了合并运算符,并不保证元素的顺序将被保留.

我的结论是否正确？文档没有涵盖我,并且不存在关于此行为的讨论.

我有一个房间查询：

@Query("SELECT * FROM classes WHERE _id IN(:values) ORDER BY date DESC")
fun getClassesByIds(values: List<Int>): Maybe<List<YClass>>

compositeDisposable.add(viewModel.getClassesById(classesTaken)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .doOnSuccess { list ->
                        list.forEach {

                            setupInfo(student)
                        }
                    }
                    .doOnError { it.printStackTrace() }
                    .subscribe())

当我从活动导航回来时，结果为空应用程序崩溃。实际上崩溃是在执行另一个查询时发生的。即使没有任何问题，第二个查询也会出错。

引起：android.arch.persistence.room.EmptyResultSetException：查询返回空结果集：SELECT * FROM classes WHERE _id IN() ORDER BY date DESC

03-06 00:07:53.476 31132-31199/? I/InputDispatcher: 窗口 'Window{135734f u0 com.labfoodandfriends.nikitagudkovs.jlog/com.labfoodandfriends.nikitagudkovs.jlog.activity.student_and_teacher.StudentOverview}' 花了 4249.1ms 处理最后一个输入事件：KeyEvent(deviceId) =0x00000101，动作=1，标志=0x00000048，keyCode=4，scanCode=0，metaState=0x00000000，repeatCount=0），policyFlags=0x6b000002 03-06 00:07:53.612/1373.612-137d I/zygote64：编译器分配了 6MB 来编译 void android.view.ViewRootImpl.performTraversals() 03-06 00:07:53.661 13731-13736/com.labfoodandfriends.nikitagudkovs.jlog I/zygote64：代码=全码缓存集合502KB, data=339KB 03-06 00:07:53.662 13731-13736/com.labfoodandfriends.nikitagudkovs.jlog I/zygote64：代码缓存收集后，
在 io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704) 03-06 00:07:55.875 13731-13731/com.labfoodandfriends.nikitagudkovs.jlog W/
System.reerr …

我可以看到RX适用于Android和UI事件处理.我很难看到RX在后端提供什么好处.

RX Java是为后端处理而设计的,还是这个概念已经走得太远了？

我是android的新手,我有一个场景,我想从多个api获取数据.假设api_a,api_b,api_c,api_d.这些api彼此独立,但我想在混合Recycler View(水平和垂直)中显示来自这些api的数据.所以我想以这样的方式进行这些api调用,以便我可以一次获取每个api数据,以便我可以在回收器视图中显示.我已经使用了改造2,但为此我必须逐个链接它们,这是非常冗长的,我认为这不是一个可行的方法.我对RX JAVA知之甚少,但我只知道如何一次发出一个请求.请帮忙

我使用 mergeLatest() 来组合 3 个可观察数据流。所有这些组合在一起，以便同时显示 UI 中的所有数据。现在，有一种情况，其中一个可观察量不会发出任何内容，因为获取的数据可能为空。

是否有 RxJava 运算符让订阅者知道不会因为空数据而发出任何信号？

编辑

private fun retrieveData() {
    Observable.combineLatest(getCurrentUser.execute(), getLatestGoal.execute(), getLatestLog.execute(),
            Function3<User, Goal, Log, PersonalViewModel> { user, goal, log -> mapToViewModel(user, goal, log) })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnSubscribe { /*todo: animation*/ }
            .doOnNext { view.setViewModel(it) }
            .doOnComplete { view.stopLoading() }
            .doOnError { /*todo: error message*/ }
            .subscribe()
}

第三个流：当用户有 nog 日志时，getLatestLog.execute() 不发出任何内容。当该流不发出时，整个视图将不可见。

数据是从 FireBase 实时数据库中获取的。ChildEventListener 有一个如下所示的方法：

override fun onChildAdded(dataSnapshot: DataSnapshot?, p1: String?) {
                val log = dataSnapshot?.getValue(Log::class.java)
                log?.let { subscriber.onNext(it) }
                subscriber.onComplete()
                firebaseDatabase.reference.removeEventListener(this)
            }

嗨,我是RxJava和Kotlin的新手,我放弃了一些关于它的概念.

我有这样的"api":

interface VehiclesService {
    @GET("/vehicles/")
    fun getVehicles(): Single<List<Vehicle>>
}

然后我创建了改造客户端等.像这样:

var retrofit = RetrofitClient().getInstance()
vehiclesAPI = retrofit!!.create(VehiclesService ::class.java)

最后我打电话:

private fun fetchData() {
        compositeDisposable.add(vehiclesAPI .getVehicles()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe { vehicles -> displayData(vehicles) }
        )
    }

这是我尝试启动时出现错误的地方:

The exception was not handled due to missing onError handler in the subscribe() method call

我知道错误是非常明确的.所以我知道缺少什么,但我不知道的是如何处理这个错误.

我尝试添加:.doOnError { error -> Log.d("MainClass",error.message) }但仍然告诉相同的错误消息.

鉴于 Hystrix 进入维护模式，我一直致力于将（相当大的）代码库迁移到 Resilience4j。

我在 Hystrix 中大量使用以下模式：

new HystrixCommand<SomeReturnValue>(DependencyKeys.DEPENDENCY) {
    @Override
    protected SomeReturnValue run() {
        return someExpensiveCall();
    }
}
    .observe()

我想用 Resilience4j 复制 Hystrix 的一些功能。

到目前为止，我有以下语法来连接外部调用：

resilience.single(DependencyKeys.DEPENDENCY, this::someExpensiveCall);

其中Resilience类提供single方法：

public <T> Single<T> single(ResilienceKey key, Callable<T> callable) {
    return Completable.complete()
            .subscribeOn(Schedulers.computation())
            .observeOn(configuration.scheduler(key))
            .andThen(Single.defer(() -> Single.fromCallable(callable)
                    .lift(CircuitBreakerOperator.of(configuration.circuitBreaker(key)))
                    .lift(RateLimiterOperator.of(configuration.rateLimiter(key)))
                    .lift(BulkheadOperator.of(configuration.bulkhead(key)))
            ))
            .observeOn(Schedulers.computation());
}

在断路和在不同线程池上运行代码方面，这看起来如何更好地类似于 Hystrix，但以更理智的方式。我真的不喜欢用 just 来启动链，这样我就可以在实际的可调用对象被包装之前Completable.complete()强制使用 a 。observeOn