RxJava中如何从每个GroupedObservable中获取单个元素

场景还原 #

定义模型类：

class SomeModel {
   int importantField1;
   int mainData;

   // 构造方法
   public SomeModel(int importantField1, int mainData) {
       this.importantField1 = importantField1;
       this.mainData = mainData;
   }
}

构造测试数据（包含重复元素）：

List<SomeModel> dataList = new ArrayList<>();
dataList.add(new SomeModel(3, 1));
dataList.add(new SomeModel(3, 1));
dataList.add(new SomeModel(2, 1));

需求：先按importantField1分组，每个分组仅保留一个元素；再按mainData分组，每个分组同样仅保留一个元素，最终得到去重后的结果列表。

原有代码问题分析 #

原有代码逻辑错误导致结果仍有重复：

List<SomeModel> resultList = Observable.fromIterable(dataList)
    .sorted((s1, s2) -> Long.compare(s2.importantField1, s1.importantField1))
    .groupBy(s -> s.importantField1)
    .firstElement() // 错误：仅获取第一个分组，而非遍历所有分组
    .flatMapObservable(item -> item)
    .groupBy(s -> s.mainData)
    .flatMap(groupedItem -> groupedItem.firstElement().toObservable())
    .toList()
    .blockingGet();

问题核心：groupBy后调用firstElement()会直接只拿第一个分组，后续所有操作都基于这个分组内的元素处理，导致后续按mainData分组时，同一个mainData下仍有重复元素。

另外尝试的Stack Overflow写法报错：

.flatMap(grp -> grp.publish(o -> o.first().concatWith(o.ignoreElements())))

报错原因：ignoreElements()返回Completable类型，而concatWith()要求传入Observable类型，两者类型不兼容，且这个写法本身属于冗余操作，完全没必要用在当前场景。

正确实现方案 #

直接对每个GroupedObservable提取首个元素，不需要额外复杂操作：

List<SomeModel> resultList = Observable.fromIterable(dataList)
    .sorted((s1, s2) -> Long.compare(s2.importantField1, s1.importantField1))
    // 按importantField1分组，每个分组取第一个元素
    .groupBy(s -> s.importantField1)
    .flatMap(grouped -> grouped.firstElement().toObservable())
    // 再按mainData分组，每个分组取第一个元素
    .groupBy(s -> s.mainData)
    .flatMap(grouped -> grouped.firstElement().toObservable())
    .toList()
    .blockingGet();

逻辑说明：

1. groupBy(s -> s.importantField1)将数据按importantField1分成多个分组
2. flatMap(grouped -> grouped.firstElement().toObservable())遍历每个分组，仅取分组内第一个元素
3. 后续按mainData分组并提取首个元素的逻辑同理，最终得到每个分组仅保留一个元素的结果列表

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