如何区分Flowable初始加载与后续单个变更事件？

嘿，这个场景我在实际项目里太熟悉了！当初用RxJava2配合本地数据库做列表展示的时候，也纠结过怎么区分全量初始加载和后续的单个条目更新。给你分享几个靠谱的解决方案，按需选就行：

这种方法通过自定义事件类型，把初始加载和单个变更的事件明确区分开，代码可读性和维护性都很高，尤其是Kotlin的密封类简直为这种场景量身定做。

Kotlin 实现示例 #

// 定义密封类区分事件类型
sealed class DataEvent {
    data class InitialLoad(val items: List<Item>) : DataEvent()
    data class ItemUpdate(val item: Item) : DataEvent()
}

// 改造你的Flowable，合并初始加载和更新事件
fun getItemsFlowable(): Flowable<DataEvent> {
    // 初始加载：从数据库获取全量数据，包装成InitialLoad事件
    val initialLoadFlowable = database.getAllItems()
        .firstOrError()
        .map { DataEvent.InitialLoad(it) }
        .toFlowable()
    
    // 单个变更：监听数据库的单条更新，包装成ItemUpdate事件
    val itemUpdatesFlowable = database.getItemUpdates()
        .map { DataEvent.ItemUpdate(it) }
    
    // 先发射初始加载事件，再持续接收更新事件
    return Flowable.concat(initialLoadFlowable, itemUpdatesFlowable)
}

订阅时直接判断事件类型处理：

getItemsFlowable()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ event ->
        when(event) {
            is DataEvent.InitialLoad -> {
                // 初始加载：直接替换RecyclerView的全部数据
                adapter.submitList(event.items)
            }
            is DataEvent.ItemUpdate -> {
                // 单个变更：只更新列表中对应的条目
                adapter.updateItem(event.item)
            }
        }
    }, { error ->
        // 处理错误逻辑
    })

Java 实现示例 #

如果用Java，可以用抽象类+子类的方式定义事件：

abstract class DataEvent {}

class InitialLoad extends DataEvent {
    private List<Item> items;
    public InitialLoad(List<Item> items) { this.items = items; }
    public List<Item> getItems() { return items; }
}

class ItemUpdate extends DataEvent {
    private Item item;
    public ItemUpdate(Item item) { this.item = item; }
    public Item getItem() { return item; }
}

// 改造Flowable
Flowable<DataEvent> getItemsFlowable() {
    Flowable<DataEvent> initialLoad = database.getAllItems()
            .firstOrError()
            .map(InitialLoad::new)
            .toFlowable();

    Flowable<DataEvent> itemUpdates = database.getItemUpdates()
            .map(ItemUpdate::new);

    return Flowable.concat(initialLoad, itemUpdates);
}

订阅时通过instanceof判断：

getItemsFlowable()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(dataEvent -> {
            if (dataEvent instanceof InitialLoad) {
                adapter.submitList(((InitialLoad) dataEvent).getItems());
            } else if (dataEvent instanceof ItemUpdate) {
                adapter.updateItem(((ItemUpdate) dataEvent).getItem());
            }
        }, throwable -> {
            // 错误处理
        });

如果你的原始Flowable首次发射全量列表，之后持续发射单个条目，可以直接用take(1)获取初始加载的事件，skip(1)获取后续的所有更新事件：

// 假设原始Flowable：首次发射List<Item>，之后发射单个Item
Flowable<Object> originalFlowable = ...;

// 拆分初始加载和更新事件
Flowable<List<Item>> initialLoad = originalFlowable
    .take(1)
    .cast(List.class)
    .map(items -> (List<Item>) items);

Flowable<Item> itemUpdates = originalFlowable
    .skip(1)
    .cast(Item.class);

// 合并订阅处理
Disposable disposable = Flowable.merge(
    initialLoad.map(items -> Pair.create("initial", items)),
    itemUpdates.map(item -> Pair.create("update", item))
)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(pair -> {
    if ("initial".equals(pair.first)) {
        adapter.submitList((List<Item>) pair.second);
    } else {
        adapter.updateItem((Item) pair.second);
    }
}, error -> {
    // 错误处理
});

⚠️ 注意：这个方案依赖事件的发射顺序，必须保证第一个事件是全量加载，后续是单个变更，否则会出错。

如果你的Flowable每次数据变化都会发射全量列表（比如Room的@Query返回的Flowable<List<Item>>），可以用一个布尔标记位判断是否是首次加载：

private boolean isFirstLoad = true;

originalFlowable
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(items -> {
        if (isFirstLoad) {
            // 首次加载：直接替换全部数据
            adapter.submitList(items);
            isFirstLoad = false;
        } else {
            // 后续更新：用DiffUtil计算差异，只更新变化的条目
            DiffUtil.DiffResult diffResult = DiffUtil.calculateDiff(
                new ItemDiffCallback(adapter.getCurrentList(), items)
            );
            adapter.submitList(items);
            diffResult.dispatchUpdatesTo(adapter);
        }
    }, error -> {
        // 错误处理
    });

这个方案实现简单，适合不需要区分单个变更、只需要优化列表刷新性能的场景。

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