RxJava2 flatMap 创建重复事件

Question

我在 RxJava2 上相对较新，我遇到了一些奇怪的行为，所以我很可能以错误的方式使用该工具。

这是一个相当大的项目，但我已将下面的 sn-p 分离为最小可重现代码：

Observable
  .interval(333, TimeUnit.MILLISECONDS)
  .flatMap(new Function<Long, ObservableSource<Integer>>() {
    private Subject<Integer> s = PublishSubject.create();
    private int val = 0;

    @Override public ObservableSource<Integer> apply(Long aLong) throws Exception {
      val++;
      s.onNext(val);
      return s;
      }
    })
  .subscribe(new Consumer<Integer>() {
    @Override public void accept(Integer integer) throws Exception {
      Log.w("value: %s", integer);
     }
  });

此代码使用.interval 模拟来自我的rx 流的事件，flatMap 接收这些事件“进行一些处理”并使用Subject 将结果推送到流中。

流是一个持续的过程，将有几个事件。

这个最小代码很愚蠢，因为我只推送 apply 回调，但在实际情况下，有几个可能的时刻可能会发生推送，并且在 apply 期间接收到的事件数量不是将通过主题发送的相同金额。

我希望通过这段代码看到的是：

value: 2  // 1 got skipped because onNext is called before there's a subscriber.
value: 3
value: 4
value: 5
value: 6 ... etc

我实际得到的是：

value: 2
value: 3
value: 3 // 3 twice
value: 4
value: 4
value: 4 // 4 repeated 3 times
value: 5
value: 5
value: 5
value: 5 // 5 repeated 4 times
value: 6
value: 6
value: 6
value: 6
value: 6 // 6 repeated 5 times
 ... etc

我也试过有一个Observable<Integer> o = s.share(); 并返回它，或者直接返回s.share(); 得到相同的结果。

我有点理解为什么会这样。 ObservableSource 再次被订阅 n 再次 n ，因此每个循环都有更多事件。

问题：

如何实现我的预期行为？

（如果我的预期行为不清楚，请在 cmets 上询问更多信息）

Answer 1

您的PublishSubject 订阅了多次，每个项目从interval() 订阅一次。

编辑：您每次都需要传入一个新的PublishSubject（如果您想保留第一个/最后一个发射，请切换到BehaviorSubject）；将其传递给长时间运行的进程，并确保在长时间运行的进程完成时正确调用其onComplete。

Answer 2

编辑

在最近的 cmets 之后，我可以想出这种解决方案：

class MyBluetoothClient {
  private PublishSubject<BTLEEvent> statusPublishSubject = PublishSubject.create()

  public Observable<BTLEEvent> getEventObservable() {
    return statusPublishSubject
  }

  private void publishEvent(BTLEEvent event) {
    statusPublishSubject.onNext(event)
  }

  public void doStuff1() {
    // do something that returns:
    publishEvent(BTLEEvent.someEvent1)
  }

  public void doStuff2() {
    // do something else that eventually yields
    publishEvent(BTLEEvent.someEvent2)
  }
}

而你以这种方式使用它：

MyBluetoothClient client = MyBluetoothClient()
client
  .getEventObservable()
  .subscribe( /* */ )

///

client.doStuff1()

/// 

client.doStuff2

原答案

这样可以吗？

Observable
  .interval(333, TimeUnit.MILLISECONDS)
  .flatMap(new Function<Long, ObservableSource<Integer>>() {
    private int val = 0;

    @Override public ObservableSource<Integer> apply(Long aLong) throws Exception {
      val++;
      return Observable.just(val);
      }
    })
  .subscribe(new Consumer<Integer>() {
    @Override public void accept(Integer integer) throws Exception {
      Log.w("value: %s", integer);
     }
  });

Answer 3

这就是我想出的答案。我会将@Tassos 的答案标记为正确，因为他指出我在正确的道路上。

首先我需要一个CachedSubject（一个在没有观察者时缓存项目并在观察者连接后立即调度它们的主题），这是确保来自apply 内部的排放真正通过的必要条件。该类主要包含PublishSubject。

class CachedSubject<T> extends Subject<T> {

        private PublishSubject<T> publishSubject = PublishSubject.create();
        private Queue<T> cache = new ConcurrentLinkedQueue<>();

        @Override public boolean hasObservers() {
            return publishSubject.hasObservers();
        }

        @Override public boolean hasThrowable() {
            return publishSubject.hasThrowable();
        }

        @Override public boolean hasComplete() {
            return publishSubject.hasComplete();
        }

        @Override public Throwable getThrowable() {
            return publishSubject.getThrowable();
        }

        @Override protected void subscribeActual(Observer<? super T> observer) {
            while (cache.size() > 0) {
                observer.onNext(cache.remove());
            }
            publishSubject.subscribeActual(observer);
        }

        @Override public void onSubscribe(Disposable d) {
            publishSubject.onSubscribe(d);
        }

        @Override public void onNext(T t) {
            if (hasObservers()) {
                publishSubject.onNext(t);
            } else {
                cache.add(t);
            }
        }

        @Override public void onError(Throwable e) {
            publishSubject.onError(e);
        }

        @Override public void onComplete() {
            publishSubject.onComplete();
        }
    }

然后我将这个类与switchMap 一起使用：

Observable
   .interval(1000, TimeUnit.MILLISECONDS)
   .switchMap(new Function<Long, ObservableSource<Integer>>() {

      private Subject<Integer> s = new CachedSubject<>();
      private int val = 0;

      @Override public ObservableSource<Integer> apply(Long aLong) throws Exception {
         val++;
         s.onNext(val);
         return s;
      }
   })
   .subscribe(new Consumer<Integer>() {
      @Override public void accept(Integer integer) throws Exception {
         Log.w("value: %s", integer);
      }
   });

这有效地允许我在 apply<T t> 方法上接收任意数量的事件，并且只有 1 个 Consumer 订阅它，从它接收所有事件。