具有多个订阅者和事件的RxJava并发

Question

我正在寻找一种将多个订阅者附加到RxJava可观察流的方法，每个订阅者异步处理发出的事件。

我第一次尝试使用。flatMap（），但这似乎对任何后续订阅服务器都不起作用。所有订阅服务器都在同一线程上处理事件。

.flatMap(s -> Observable.just(s).subscribeOn(Schedulers.newThread()))

最终工作的是通过每次创建一个新的可观察的来消耗新线程中的每个事件：

Observable.from(Arrays.asList(new String[]{"1", "2", "3"}))
            .subscribe(j -> {
                Observable.just(j)
                        .subscribeOn(Schedulers.newThread())
                        .subscribe(i -> {
                            try {
                                Thread.sleep(ThreadLocalRandom.current().nextInt(100, 500));
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            System.out.println("s1=>" + Thread.currentThread().getName() + "=>" + i);
                        });
            });

输出：

s1=>RxNewThreadScheduler-1=>1
s1=>RxNewThreadScheduler-2=>2
s1=>RxNewThreadScheduler-3=>3

以及多个订阅者的最终结果：

ConnectableObservable<String> e = Observable.from(Arrays.asList(new String[]{"1", "2", "3"}))
            .publish();

    e.subscribe(j -> {
        Observable.just(j)
                .subscribeOn(Schedulers.newThread())
                .subscribe(i -> {
                    try {
                        Thread.sleep(ThreadLocalRandom.current().nextInt(100, 500));
                    } catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                    System.out.println("s1=>" + Thread.currentThread().getName() + "=>" + i);
                });
    });

    e.subscribe(j -> {
        Observable.just(j)
                .subscribeOn(Schedulers.newThread())
                .subscribe(i -> {
                    try {
                        Thread.sleep(ThreadLocalRandom.current().nextInt(100, 500));
                    } catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                    System.out.println("s2=>" + Thread.currentThread().getName() + "=>" + i);
                });
    });

    e.connect();

输出：

s2=>RxNewThreadScheduler-4=>2
s1=>RxNewThreadScheduler-1=>1
s1=>RxNewThreadScheduler-3=>2
s2=>RxNewThreadScheduler-6=>3
s2=>RxNewThreadScheduler-2=>1
s1=>RxNewThreadScheduler-5=>3

然而，这似乎有点笨拙。有没有更优雅的解决方案，或者RxJava不是一个很好的用例？

Answer 1

flatMap中的Observable上的doOnNext将产生与您相同的输出。

始终以顺序方式调用onNext（），因此在平面图之后使用doOnNext也不适用。由于同样的原因，在最终订阅中写入操作在您的案例中不起作用。

下面的代码是使用RxJava2编写的。在RxJava的版本1中，您必须在线程周围添加try-catch块。睡眠。

ConnectableObservable<String> e = Observable.just("1", "2", "3").publish();

e.flatMap(
      s -> Observable.just(s)
          .subscribeOn(Schedulers.newThread())
          .doOnNext(i -> {  // <<<<<<
              Thread.sleep(ThreadLocalRandom.current().nextInt(100, 500));
              System.out.println("s1=>" + Thread.currentThread().getName() + "=>" + i);
          }))
  .subscribe();

e.flatMap(
      s -> Observable.just(s)
          .subscribeOn(Schedulers.newThread())
          .doOnNext(i -> {  // <<<<<<
              Thread.sleep(ThreadLocalRandom.current().nextInt(100, 500));
              System.out.println("s2=>" + Thread.currentThread().getName() + "=>" + i);
          }))
  .subscribe();

e.connect();

Answer 2

如果我正确理解了rx合约，你正在试图做一些事情，这是违背它的。

让我们看看合同

RxJava Observable的约定是事件（onNext（）、onCompleted（）、onEr ror（））永远不能同时发出。换言之，单个可观察流必须始终序列化且线程安全。每个事件都可以从不同的线程发出，只要这些发出不是并发的。这意味着onNext（）不能相互离开或同时执行。如果onNext（）仍在一个线程上执行，则另一个线程无法再次开始调用它（交错）--Tomasz Nurkiewicz在RxJava反应式编程中的应用

在我看来，您试图通过在外部订阅中使用嵌套订阅来打破合同。对订阅者的onNext调用不再序列化。

为什么不将“异步”工作负载从订阅者移动到flatMap操作符并订阅新的observable：

    ConnectableObservable<String> stringObservable = Observable.from(Arrays.asList(new String[]{"1", "2", "3"}))
            .flatMap(s -> {
                return Observable.just(s).subscribeOn(Schedulers.computation());
            })
            .publish();

    stringObservable
            .flatMap(s -> {
                // do More asyncStuff depending on subscription
                return Observable.just(s).subscribeOn(Schedulers.newThread());
            })
            .subscribe(s -> {
                // use result here
            });

    stringObservable
            .subscribe(s -> {
                // use immediate result here.
            });

    stringObservable.connect();

Answer 3

使用<代码>。平面图-