我对 flatMap 如何控制它的“子”线程很感兴趣，例如下面的代码工作正常:

 private Flowable<PlcDataPackage> createIntervalPlcFlowable() {
    return Flowable.interval(1, TimeUnit.SECONDS, Schedulers.computation())
            .onBackpressureLatest()
            .parallel()
            .runOn(Schedulers.computation())
            .flatMap((Function<Long, Publisher<PlcDataPackage>>) aLong -> mDataPackageFlowable)
            .sequential();
}

此代码在被调用 128 次后停止(即 flowable 的默认 maxConcurent):

  private ConnectableFlowable<PlcDataPackage> createConnectablePlcFlowable() {
    return mPlcIntervalFlowable.onBackpressureLatest()
            .subscribeOn(Schedulers.single())
            .publish();
}

订阅:

addDisposable(mGetPlcUpdatesChanelUseCase.execute()
                              .observeOn(AndroidSchedulers.mainThread())
                              .subscribe(plcDto -> Timber.d("plcReceiver"),
                                         Timber::e));

用例:

public class GetPlcUpdatesChanelUseCase extends UseCase<PlcDto, Object> {

    private final PlcRepository mPlcRepository;

    public GetPlcUpdatesChanelUseCase(PlcRepository plcRepository) {
        mPlcRepository = plcRepository;
    }

    @Override
    public Flowable<PlcDto> buildFlowable(Optional<Object> optional) {
        return mPlcRepository.getUpdatesChannel();
    }

    @Override
    public boolean isParamsRequired() {
        return false;
    }
}

repo 方法

@Override
    public Flowable<PlcDto> getUpdatesChannel() {
        return mPlcCore.getPlcConnectableFlowable()
                .map(mPlcInfoTopPlcDtoTransformer::transform);
    }

PlcCore 方法

public ConnectableFlowable<PlcDataPackage> getPlcConnectableFlowable() {
    return mConnectableFlowable;
}

而 mConnectableFlowable 是:

mConnectableFlowable = createConnectablePlcFlowable();
        mConnectableFlowable.connect();

所以据我所知，mDataPackageFlowable 被创建一次，然后被执行，每次它为它的 child 创建新的“线程”，并且在 128 次执行之后它只会阻止所有后续执行。

所以有3个主要问题:

1) flatMap是否控制子线程？

2) 为什么它在新线程上执行每个新的“请求”？(也许不是，然后告诉我)

3) 在什么情况下我们会失去对子线程的控制。

免责声明:英语是我的第二语言，如果有什么不清楚的地方可以问我，我会尽量补充说明。

 private Flowable<PlcDataPackage> createIntervalPlcFlowable() {
    return Flowable.interval(1, TimeUnit.SECONDS, Schedulers.computation())
            .onBackpressureLatest()
            .parallel()
            .runOn(Schedulers.computation())
            .sequental()

这个组合不起作用，它实际上去除了 128 次 flatMap 调用限制，但没有清除导致内存泄漏和 OOM 异常的旧 innersubscription。请改用某种 map 。

最佳答案

观察者链需要订阅才能正常工作。当您使用 interval() 生成数据时，您提供了一个“热”可观察对象，它会自行发出值。 “冷”可观察对象仅在订阅发生时才会发出值。

128 是 flatMap() 在暂停之前缓冲的条目数。如果有订阅，则 flatMap() 将向下游发出内部可观察对象生成的值，并且不会停止。

flatMap() 本身并不在特定的调度程序上运行。这意味着它不会在特定线程上操纵其订阅。如果您想控制 flatMap() 调用的可观察对象中正在完成的工作，那么您可以使用显式调度:

observable
  .flatMap( value -> fun(value).subscribeOn( myScheduler ) )
  .subscribe();

myScheduler 例如可能是 Schedulers.io()，它会在需要时创建线程。或者，它可以是您提供的具有固定数量线程的 Executor。我经常使用只分配了一个或两个或 48 个线程的 Executor 来控制 flatMap() 的扇出。

您还可以向 flatMap() 提供一个并行参数，告诉它要维护的最大订阅数。当 flatMap() 达到最大值时，它将缓冲请求，直到它订阅的观察者链完成。

parallel() 运算符做类似的事情，但它将传入事件分开，在单独的线程上发出它们。同样，javadoc 具有出色的描述和精美的图片。

总是有可能失去对线程的控制。当您使用 RxJava 运算符时，请阅读它的文档。您需要了解两个方面。第一个区域是运算符(operator)在哪个调度程序上工作。如果它说它不在特定的调度程序上运行，那么它不会直接影响线程的选择或线程的使用方式。如果它声明它使用特定的调度程序，那么您需要了解该调度程序的工作原理；总会有另一个版本的运算符(operator)允许您提供自己选择的调度程序。

您必须了解的第二个方面是背压。您需要了解背压的含义及其应用方式。每当您跨越线程边界时，这一点尤其重要，例如使用 observeOn() 或 subscribeOn()。

关于android - flatMap 是如何深入工作的？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/53555124/