如何解决Akka Streams过滤自定义类型Source时的编译报错问题

错误原因 #

你定义的Source元素类型为List<RData>，流中每一个下发的元素都是完整的List集合，而非单个RData对象。调用filter算子时，入参x是List<RData>类型，Java的List本身没有getId()方法，因此触发编译错误。
filter算子的作用是对流中每一个完整元素做断言校验，不会自动遍历元素内部的集合结构，需要先处理集合结构再做过滤。

解决方案 #

根据你的实际业务需求，有两种常见实现方式：

方案1：保留List结构，过滤List内部元素 #

如果你需要流输出的元素依然是List<RData>类型，仅剔除List内不符合条件的RData，可以通过map算子对每个List做内部过滤：

final Source<List<RData>, NotUsed> filteredSource = source.map(rDataList -> 
    rDataList.stream()
        .filter(rData -> rData.getId().equals("1"))
        .toList()
);

方案2：拆分集合为单元素流后过滤 #

如果你需要流输出的元素是单个RData对象，可以用Akka Streams提供的mapConcat算子先把List拆分为单个元素下发，再执行过滤操作：

final Source<RData, NotUsed> filteredSingleSource = source
    // 拆分List为单个RData元素发射
    .mapConcat(rDataList -> rDataList)
    // 此时流元素为RData类型，可直接调用getId过滤
    .filter(rData -> rData.getId().equals("1"));

另外注意你当前的builderFunction每次调用都会向静态全局Listls追加元素，重复调用会导致数据冗余，如果需要每次生成全新的数据集，建议将ls改为builderFunction内部的局部变量。

内容的提问来源于stack exchange，提问作者blue-sky