0

我正在使用来自不同外部源/订阅包装的事件来使用不同的 Flowable。来源无关紧要，因为我可以通过简单的循环重现该问题。我有 ：

  • 不同的FlowableEmitter（3个足以重现）
  • 发射器的单线程（下面是主线程）
  • 订阅者的单线程 (newSingleThreadExecutor)

这是一个简单的代码重现

    final ExecutorService executor = Executors.newSingleThreadExecutor();
    final Scheduler scheduler = Schedulers.from(executor);

    List<FlowableEmitter<Long>> emitterList = new ArrayList<>();
    for (int i=0; i<3; ++i) {
        final int finalI = i;
        Flowable.create( new FlowableOnSubscribe<Long>(){
            @Override
            public void subscribe(FlowableEmitter<Long> emitter) {
                emitterList.add(emitter);
            }
        },  BackpressureStrategy.MISSING)
                .observeOn(scheduler)
                .subscribe(
                        val -> System.out.println(
                            "[" +Thread.currentThread().getName()
                            + "] Flow:" + finalI 
                            + " > " + Long.toString(val)));
    }

    long state = 0;
    for (int i=0; i<5; ++i) {
        for (FlowableEmitter<Long> emitter: emitterList){
            emitter.onNext(++state);
        }
    }

    executor.shutdown();

我的问题是事件的使用顺序与它们发出的顺序不同。如果我删除了 observeOn(scheduler) 它工作正常，但我需要在不同的线程上发射和订阅。我还测试了不同的 BackpressureStrategy 并没有帮助。
任何线索都可以按顺序订阅/打印所有号码（1,2,3,4,5...14,15），而不是我下面的

[pool-1-thread-1] Flow:0 > 1
[pool-1-thread-1] Flow:0 > 4
[pool-1-thread-1] Flow:0 > 7
[pool-1-thread-1] Flow:0 > 10
[pool-1-thread-1] Flow:0 > 13
[pool-1-thread-1] Flow:1 > 2
[pool-1-thread-1] Flow:1 > 5
[pool-1-thread-1] Flow:1 > 8
[pool-1-thread-1] Flow:1 > 11
[pool-1-thread-1] Flow:1 > 14
[pool-1-thread-1] Flow:2 > 3
[pool-1-thread-1] Flow:2 > 6
[pool-1-thread-1] Flow:2 > 9
[pool-1-thread-1] Flow:2 > 12
[pool-1-thread-1] Flow:2 > 15

如果有问题，我正在使用 rx-java 2.2.5 和 Java 8。

4

1 回答 1

0

observeOn不公平，如果有更多工作要做，可能会拥抱发射线程。通过循环发射，一个源可能准备发射更多，因此您无法获得完美的交错。另请注意，通过使用MISSING策略，您很容易出现背压异常。从扩展项目中尝试requestObserveOn ：

final ExecutorService executor = Executors.newSingleThreadExecutor();
final Scheduler scheduler = Schedulers.from(executor);

List<FlowableEmitter<Long>> emitterList = new ArrayList<>();
for (int i=0; i<3; ++i) {
    final int finalI = i;
    Flowable.create( new FlowableOnSubscribe<Long>(){
        @Override
        public void subscribe(FlowableEmitter<Long> emitter) {
            emitterList.add(emitter);
        }
    },  BackpressureStrategy.BUFFER)
            // ---------------------------------------------------------
            .compose(FlowableTransformers.requestObserveOn(scheduler))
            // ---------------------------------------------------------
            .subscribe(
                    val -> System.out.println(
                        "[" +Thread.currentThread().getName()
                        + "] Flow:" + finalI 
                        + " > " + Long.toString(val)));
}

long state = 0;
for (int i=0; i<5; ++i) {
    for (FlowableEmitter<Long> emitter: emitterList){
        emitter.onNext(++state);
    }
}

executor.shutdown();
于 2019-03-04T14:28:25.700 回答