代码之家 › 专栏 › 技术社区 › j2emanue

RxJava2可连接可观察-重播未发送所有以前的项目

rx-java2

j2emanue · 技术社区 · 6 年前

 List<Integer> list = new ArrayList<Integer>();
    for(int j=1;j<=3;j++)
        list.add(j);


    Observable<Integer> observable = Observable.fromIterable(list)
    .replay()
    .autoConnect();



    observable.subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            Log.v("consumer1:", ""+integer);
        }
    });


    observable.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
    Log.v("consumer2:", ""+integer);

}
});



    observable.subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            Log.v("consumer3:", ""+integer);
        }
    });

当我运行上述代码时,我得到以下输出:

consumer1:: 1
consumer1:: 2
consumer1:: 3
consumer2:: 1
consumer2:: 2
consumer2:: 3
consumer3:: 1
consumer3:: 2
consumer3:: 3

我希望replay能够真正“重播”之前发生的所有历史。所以我希望能发出之前发生的所有流。特别是这是我期望的输出:

//first time nothing to replay so just do the work
consumer1:: 1
consumer1:: 2
consumer1:: 3

//replay consumer1 stream:,
consumer2:: 1
consumer2:: 2
consumer2:: 3
//already replayed now do the work
consumer2:: 1
consumer2:: 2
consumer2:: 3

//replay consumer1 stream:
consumer3:: 1
consumer3:: 2
consumer3:: 3

//replay consumer2 stream:
consumer3:: 1
consumer3:: 2
consumer3:: 3
//now do the work
consumer3:: 1
consumer3:: 2
consumer3:: 3

1 回复 | 直到 6 年前

akarnokd 6 年前

不知道你为什么想要这种异国情调的行为,但你可以不断重复:

List<Integer> list = new ArrayList<Integer>();
for (int j = 1; j <= 3; j++) {
    list.add(j);
}


AtomicInteger count = new AtomicInteger();

Observable<Integer> observable = 
    Observable.defer(() -> {
        Observable.fromIterable(list)
        .replay()
        .autoConnect()
        .repeat(count.incrementAndGet());
    });

observable.subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        Log.v("consumer1:", ""+integer);
    }
});

observable.subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        Log.v("consumer2:", "" + integer);
    }
});

observable.subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        Log.v("consumer3:", ""+integer);
    }
});

推荐文章

Gissipi_453 · 在RxJava 2中的可观察对象中发出字符串数组时出现“不兼容类型”错误

6 年前

Damn Vegetables · 当N个事件中的任何一个发生时执行,但使用所有其他事件的最后结果

6 年前

Damn Vegetables · 如何压缩(s1,o1),其中s1只发生一次,o1发生多次?

6 年前

CaHa · 如何在rxjava中等待改装响应

6 年前

noloman amram99 · Android和Room:插入作品,查询不

6 年前

Martin · 可观察到。具有状态的生成器,其中异步更新新状态

6 年前

Victor Dmitrienko · RxJava解Bounce运算符取决于消息

6 年前

idunnololz adsion · 调用dispose()之前是否需要检查isDisposed()?

7 年前

Josh Laird · 轮询第二个URL,直到JSON返回预期参数

7 年前

n0shadow · RxJava:为一个完整表列表指定最大并发性

7 年前