RxJava：如何为replay()操作符实现自定义过期避免内存泄漏？

你可以通过**scan操作符维护实体状态**的方式替代分组+replay的方案，既满足“订阅后获取每个活跃实体最新Snapshot+后续新Snapshot”的需求，又能自动清理超时实体，避免内存泄漏，无需手动管理缓存。

实现代码 #

首先定义一个内部类来存储每个实体的最新Snapshot和时间戳，用于判断是否超时：

private static class SnapshotEntry {
    final Snapshot snapshot;
    final long timestamp;

    SnapshotEntry(Snapshot snapshot) {
        this.snapshot = snapshot;
        this.timestamp = System.currentTimeMillis();
    }
}

然后基于源流构建UI所需的热Observable：

// 先将源流转为热流，确保所有订阅共享同一数据源
Observable<Snapshot> hotSnapshots = getContinuousSnapshotsStream()
        .publish()
        .autoConnect();

// 用scan维护每个实体的最新状态，自动清理超时（1天）的实体
Observable<Map<String, Snapshot>> activeSnapshotsState = hotSnapshots
        .scan(new ConcurrentHashMap<>(), (currentMap, newSnapshot) -> {
            // 更新当前实体的最新Snapshot和时间戳
            currentMap.put(newSnapshot.getId(), new SnapshotEntry(newSnapshot));
            
            // 移除超过1天未更新的实体
            long cutoffTime = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(1);
            currentMap.entrySet().removeIf(entry -> entry.getValue().timestamp < cutoffTime);
            
            // 返回新的Map副本，避免并发修改问题
            return new HashMap<>(currentMap);
        })
        .map(entryMap -> {
            // 转换为只包含Snapshot的Map，去掉时间戳
            Map<String, Snapshot> resultMap = new HashMap<>();
            entryMap.forEach((id, entry) -> resultMap.put(id, entry.snapshot));
            return resultMap;
        })
        .distinctUntilChanged(); // 仅当Map内容变化时才发射，避免冗余更新

// 转换为UI表格所需的Snapshot流：新订阅时先发射所有活跃实体的最新Snapshot，后续发射新的Snapshot
Observable<Snapshot> uiSnapshotStream = activeSnapshotsState
        .flatMapIterable(Map::values)
        // 避免同一实体的相同Snapshot重复发射
        .distinctUntilChanged(s -> s.getId(), (s1, s2) -> s1.equals(s2));

为什么这个方案可行？ #

1. 自动清理超时实体：在scan的状态更新逻辑中，会定期移除超过1天未更新的实体，不会保留无效引用，彻底解决内存泄漏问题。
2. 热Observable特性：通过publish().autoConnect()将源流转为热流，确保所有UI订阅共享同一数据源，不会重复消费。
3. 满足UI需求：新订阅uiSnapshotStream时，会先拿到当前所有活跃实体的最新Snapshot（用于表格初始化），后续会自动接收新的Snapshot更新。

如果你坚持使用分组方案 #

原方案的内存泄漏根源是replay()会保留所有发射过的分组Observable（即使分组已超时完成）。RxJava没有直接的操作符可以让replay()自动移除已完成的Observable，但可以通过flatMap的特性间接实现：

Observable<Snapshot> uiStream = getContinuousSnapshotsStream()
        .publish()
        .autoConnect()
        .groupBy(Snapshot::getId)
        .flatMap(group -> 
            group.timeout(1, TimeUnit.DAYS)
                 .onErrorComplete()
                 .replay(1)
                 .autoConnect()
        );

但这个方案的缺点是：新订阅只能获取订阅后新增的实体和已有实体的后续Snapshot，无法拿到订阅时已有活跃实体的最新Snapshot，不符合UI表格初始化的需求。因此更推荐前面的scan状态维护方案。

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