Java RX中带Buffer与多值更新的Observable问题求助

嘿，刚上手RxJava时对Buffer操作符的用法摸不准太正常了，我来帮你理清楚怎么实现带Buffer的Observable来处理多值更新~

首先先看你现有代码里的小细节：你用的Observable.timer只会延迟指定时间后发射一次事件就结束了，如果要持续产生事件来测试Buffer的效果，建议改成Observable.interval（每隔固定时间发射一次）或者给timer加上repeat()让它重复触发。

下面针对不同的缓冲需求给你几种常见的实现方式：

1. 按时间间隔缓冲事件 #

如果想每隔一段时间，把这段时间内产生的所有事件打包处理，比如每5秒处理一次这段时间内的timer/interval事件：

// 用interval持续每秒发射一个事件
Observable<Long> observable1 = Observable.interval(1, TimeUnit.SECONDS);

observable1.buffer(5, TimeUnit.SECONDS) // 每5秒缓冲一次这段时间内的所有事件
    .subscribe(new Subscriber<List<Long>>() {
        @Override
        public void onNext(List<Long> bufferedEvents) {
            // 这里处理缓冲后的多值列表，比如打印数量或者逐个处理
            System.out.println("收到缓冲事件，共" + bufferedEvents.size() + "个：" + bufferedEvents);
        }

        @Override
        public void onCompleted() {}

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

2. 按事件数量缓冲 #

如果想攒够指定数量的事件再统一处理，比如每3个事件打包一次：

Observable<Long> observable1 = Observable.interval(1, TimeUnit.SECONDS);

observable1.buffer(3) // 每攒够3个事件就触发一次
    .subscribe(new Subscriber<List<Long>>() {
        @Override
        public void onNext(List<Long> bufferedEvents) {
            System.out.println("攒够3个事件，开始处理：" + bufferedEvents);
        }

        // 其他方法实现同上
    });

3. 用另一个Observable作为缓冲边界 #

你代码里还有个otherObservable，可以把它作为缓冲的“触发信号”——当otherObservable发射事件时，就把之前收集的所有事件打包发送：

// 先让otherObservable有实际触发逻辑，比如每10秒发射一次
Observable<Long> otherObservable = Observable.interval(10, TimeUnit.SECONDS);
Observable<Long> observable1 = Observable.interval(1, TimeUnit.SECONDS);

observable1.buffer(otherObservable)
    .subscribe(new Subscriber<List<Long>>() {
        @Override
        public void onNext(List<Long> bufferedEvents) {
            System.out.println("otherObservable触发，处理缓冲事件：" + bufferedEvents);
        }

        // 其他方法实现同上
    });

额外注意事项 #

• Buffer操作符会把事件打包成List<T>发射，所以Subscriber的泛型要对应改成List<Long>，而不是单个Long
• 如果需要同时满足“时间到”或“数量够”任一条件就触发缓冲，可以用重载方法buffer(long timespan, TimeUnit unit, int count)
• 记得在合适的时机调用unsubscribe()来避免内存泄漏

内容的提问来源于stack exchange，提问作者jpganz18