热 Observable 的 RxJava 延迟答案

【问题标题】：RxJava Delay for Hot Observables热 Observable 的 RxJava 延迟
【发布时间】：2017-11-07 11:48:11
【问题描述】：

我看到了这个问题here。这是关于实现每个发射项目的延迟。这是基于accepted answer的实现方法：

Observable.zip(Observable.range(1, 5)
    .groupBy(n -> n % 5)
    .flatMap(g -> g.toList()),
Observable.interval(50, TimeUnit.MILLISECONDS),
(obs, timer) -> obs)
.doOnNext(item -> {
  System.out.println(System.currentTimeMillis() - timeNow);
  System.out.println(item);
  System.out.println(" ");
}).toList().toBlocking().first();

在问题中，提问者专门要求了一组固定的可观察对象（Observable.range(1,5)），不幸的是这不是我想要实现的。

我也看到了这个comment。

该评论是我想要实现的。所以我的源 observable 以比间隔更慢（有时更快）的速率发出项目。并且 observable 的发射是永无止境的。

===

所以基本上我希望 hot observables 有最小的延迟。

例如，如果我想要 400 毫秒的最小延迟并且我有这种可观察的发射率：

X1-100ms延迟-X2-200ms延迟-X3-600ms延迟-X4-20000ms延迟-X5-...

我希望它屈服：

X1-400ms 延迟-X2-400ms 延迟-X3-600ms 延迟-X4-20000ms 延迟-X5-...

任何人有任何想法来实现这一目标？

【问题讨论】：

标签： java rx-java delay

【解决方案1】：

你的要求好奇怪……

我可以解决它，但不优雅。这是我的代码：

class Three<A, B, C> {
    A a;
    B b;
    C c;
    // Getter, Setter, Constructor
  }

  public static void main(String[] args) throws Exception {
    BehaviorSubject<Integer> s = BehaviorSubject.create();
    // Three = (The value, upstream comes mills, downstream emits mills)
    s.map(i -> new Three<>(i, System.currentTimeMillis(), System.currentTimeMillis()))
        .scan((a, b) -> {
          b.setC(a.getC() + Math.max(400L, b.getB() - a.getB()));
          return b;
        })
        .concatMap(i -> Observable.just(i.getA()).delay(Math.max(0, i.getC() - System.currentTimeMillis()),
            TimeUnit.MILLISECONDS))
        .subscribe(i -> System.out.println(i + "\t" + System.currentTimeMillis()));
    s.onNext(0);
    Thread.sleep(100);
    s.onNext(1);
    Thread.sleep(200);
    s.onNext(2);
    Thread.sleep(600);
    s.onNext(3);
    Thread.sleep(2000);
    s.onNext(4);
    Thread.sleep(200);
    s.onNext(5);
    Thread.sleep(800);
    s.onNext(6);
    Thread.sleep(1000);
  }

和输出

0   1510128693984
1   1510128694366 // 400ms
2   1510128694766 // 400ms
3   1510128695366 // 600ms
4   1510128697366 // 2000ms
5   1510128697766 // 400ms
6   1510128698567 // 800ms

【讨论】：