RxJava：如何实现带固定大小缓冲区的日志即时刷新？

我来给你一个完美适配需求的方案——既能保留原有的1秒定时刷新、300条容量上限逻辑，又能支持按需即时刷新，完全用RxJava原生操作符实现，不需要hack！

核心思路 #

把三个触发刷新的条件（定时、容量达标、手动触发）合并成一个Observable，用它来控制buffer的刷新时机。RxJava的buffer操作符有个重载可以接受一个"关闭信号"Observable：每当这个Observable发射事件时，就会把当前缓冲区的内容发送出去并清空，正好匹配我们的需求。

分步实现代码 #

1. 定义手动触发的Subject #

创建一个Subject来接收手动刷新指令，调用它的onNext()就能触发即时刷新：

// 泛型用Void即可，我们只关心事件触发，不关心携带的内容
PublishSubject<Void> manualFlush = PublishSubject.create();

2. 构建三个刷新触发源 #

• 定时触发：和你原有逻辑一致，每1秒发射一个刷新信号

Observable<Void> timedTrigger = Observable
    .interval(1000, TimeUnit.MILLISECONDS, Schedulers.io())
    .map(tick -> (Void) null); // 转换为Void类型，统一触发信号格式

• 容量触发：每当日志累计到300条时自动触发刷新。这里用window(300)自动分割流，比手动计数更可靠，能处理所有边界情况：

Observable<Void> capacityTrigger = logStream
    .window(300) // 每300条日志生成一个独立窗口
    .flatMap(window -> window.lastOrDefault(null) // 监听窗口结束事件
        .map(ignored -> (Void) null)); // 转换为统一的Void触发信号

• 手动触发：就是我们刚才定义的manualFlush Subject

3. 合并触发源并应用到Buffer #

把三个触发源合并成一个Observable，传给buffer操作符控制刷新：

// 合并三个触发信号：任意一个信号到来，都会立即刷新缓冲区
Observable<Void> flushTriggers = Observable.merge(
    timedTrigger,
    capacityTrigger,
    manualFlush
);

// 替换原有buffer逻辑，用合并后的触发源控制刷新时机
logStream
    .subscribeOn(Schedulers.io())
    .buffer(flushTriggers) // 关键：用触发信号决定何时发送缓冲区内容
    .subscribe(dumpLogs, errorHandler);

4. 调用按需刷新 #

当你需要即时刷新日志时，在任意线程调用即可：

// 发送手动刷新信号，缓冲区会立即清空并发送当前所有日志
manualFlush.onNext(null);

方案优势 #

• 完全保留原有定时+容量限制的核心逻辑，没有功能损失
• 手动刷新会立即发送当前缓冲区的所有日志（无论数量多少、时间是否到1秒），发送后自动清空缓冲区
• 全部使用RxJava原生操作符，稳定性高，无需自定义复杂逻辑
• 线程模型与你原有代码一致，不会引入额外线程问题

注意事项 #

• 如果logStream是多线程发射日志，无需担心线程安全：RxJava的PublishProcessor和合并操作符都已处理线程安全问题
• 应用退出时记得调用manualFlush.onComplete()（RxJava 2+）或manualFlush.onCompleted()（RxJava 1.x）释放资源

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