如何在 Stream 上重用过滤器和映射的应用程序？

Question

我有一组继承自共享类型（即GroupRecord extends Record、RequestRecord extends Record）的域对象。子类型具有特定属性（即GroupRecord::getCumulativeTime、RequestRecord::getResponseTime）。

此外，由于解析日志文件，我有一个包含混合子类型的记录列表。

List<Record> records = parseLog(...);

为了计算日志记录的统计信息，我想仅对匹配特定子类型的记录子集应用数学函数，即仅在 GroupRecords 上。因此，我想要过滤特定子类型的流。我知道我可以将filter 和map 应用于子类型

records.stream()
       .filter(GroupRecord.class::isInstance)
       .map(GroupRecord.class::cast)
       .collect(...

多次在流上应用此过滤器和强制转换（尤其是在为不同的计算多次为同一子类型执行此操作时）不仅麻烦而且会产生大量重复。

我目前的方法是使用TypeFilter

class TypeFilter<T>{

    private final Class<T> type;

    public TypeFilter(final Class<T> type) {
        this.type = type;
    }

    public Stream<T> filter(Stream<?> inStream) {
        return inStream.filter(type::isInstance).map(type::cast);
    }
}

应用于流：

TypeFilter<GroupRecord> groupFilter = new TypeFilter(GroupRecord.class); 

SomeStatsResult stats1 = groupFilter.filter(records.stream())
                                      .collect(...)
SomeStatsResult stats2 = groupFilter.filter(records.stream())
                                      .collect(...)

它有效，但我发现这种方法对于这样一个简单的任务来说有点过分。因此，我想知道，是否有更好的或最好的方法来以简洁易读的方式使用流和函数使这种行为可重用？

Answer 1

这取决于你觉得什么“更简洁易读”。我自己会争辩说，您已经实现的方式就可以了。

但是，确实有一种方法可以通过使用Stream.flatMap，以一种比你使用它的地方略短的方式来做到这一点：

static <E, T> Function<E, Stream<T>> onlyTypes(Class<T> cls) {
  return el -> cls.isInstance(el) ? Stream.of((T) el) : Stream.empty();
}

它的作用是将每个原始流元素转换为一个元素的Stream（如果该元素具有预期类型），或者转换为一个空的Stream（如果没有）。

而用途是：

records.stream()
  .flatMap(onlyTypes(GroupRecord.class))
  .forEach(...);

这种方法有明显的权衡：

• 您确实丢失了管道定义中的“过滤器”字样。这可能比原始名称更令人困惑，因此可能需要一个比 onlyTypes 更好的名称。
• Stream 对象相对重量级，创建太多可能会导致性能下降。但是你不应该相信我的话，并在高负载下分析这两种变体。

编辑：

由于问题是用稍微更笼统的术语询问重用filter 和map，我觉得这个答案也可以讨论更多抽象。因此，要概括地重用过滤器和映射，您需要以下内容：

static <E, R> Function<E, Stream<R>> filterAndMap(Predicate<? super E> filter, Function<? super E, R> mapper) {
   return e -> filter.test(e) ? Stream.of(mapper.apply(e)) : Stream.empty();
}

原来的onlyTypes 实现现在变成了：

static <E, R> Function<E, Stream<R>> onlyTypes(Class<T> cls) {
  return filterAndMap(cls::isInstance, cls::cast);
}

但是，再次进行权衡：生成的平面映射器函数现在将保存捕获的两个对象（谓词和映射器），而不是上述实现中的单个 Class 对象。这也可能是一种过度抽象的情况，但这取决于您需要该代码的位置和原因。

Answer 2

您不需要整个类来封装一段代码。用于该目的的最小代码单元将是一个方法：

public static <T> Stream<T> filter(Collection<?> source, Class<T> type) {
    return source.stream().filter(type::isInstance).map(type::cast);
}

这个方法可以作为

SomeStatsResult stats1 = filter(records, GroupRecord.class)
                            .collect(...);
SomeStatsResult stats2 = filter(records, GroupRecord.class)
                            .collect(...);

如果过滤操作并不总是链中的第一步，则可以重载该方法：

public static <T> Stream<T> filter(Collection<?> source, Class<T> type) {
    return filter(source.stream(), type);
}
public static <T> Stream<T> filter(Stream<?> stream, Class<T> type) {
    return stream.filter(type::isInstance).map(type::cast);
}

但是，如果您必须对同一类型重复此操作多次，这样做可能会有所帮助

List<GroupRecord> groupRecords = filter(records, GroupRecord.class)
                            .collect(Collectors.toList());
SomeStatsResult stats1 = groupRecords.stream().collect(...);
SomeStatsResult stats2 = groupRecords.stream().collect(...);

不仅消除了源代码中的代码重复，而且只执行一次运行时类型检查。所需额外堆空间的影响取决于实际用例。

Answer 3

什么你真正需要的是一个Collector 来收集流中特殊类型实例的所有元素。它可以轻松解决您的问题并避免过滤流两次：

List<GroupRecord> result = records.stream().collect(
      instanceOf(GroupRecord.class, Collectors.toList())
); 

SomeStatsResult stats1 = result.stream().collect(...);
SomeStatsResult stats2 = result.stream().collect(...);

AND您可以通过使用Collectors#mapping 来执行类似Stream#map 的操作，例如：

List<Integer> result = Stream.of(1, 2L, 3, 4.)
   .collect(instanceOf(Integer.class, mapping(it -> it * 2, Collectors.toList())));
               |                                                       |  
               |                                                     [2,6]
             [1,3]

WHERE您只想使用一次Stream，您可以轻松地编写最后一个Collector，如下所示：

SomeStatsResult stats = records.stream().collect(
      instanceOf(GroupRecord.class, ...)
); 

---

static <T, U extends T, A, R> Collector<T, ?, R> instanceOf(Class<U> type
        , Collector<U, A, R> downstream) {
    return new Collector<T, A, R>() {
        @Override
        public Supplier<A> supplier() {
            return downstream.supplier();
        }

        @Override
        public BiConsumer<A, T> accumulator() {
            BiConsumer<A, U> target = downstream.accumulator();
            return (result, it) -> {
                if (type.isInstance(it)) {
                    target.accept(result, type.cast(it));
                }
            };
        }

        @Override
        public BinaryOperator<A> combiner() {
            return downstream.combiner();
        }

        @Override
        public Function<A, R> finisher() {
            return downstream.finisher();
        }

        @Override
        public Set<Characteristics> characteristics() {
            return downstream.characteristics();
        }
    };
}

为什么需要组成 Collectors？

你还记得Composition over Inheritance Principle吗？你还记得单元测试中的assertThat(foo).isEqualTo(bar) 和assertThat(foo, is(bar)) 吗？

组合更加灵活，它可以在运行时重用一段代码和组合组件，这就是为什么我更喜欢@ 987654336@ 而不是fest-assert 因为它可以将所有可能的Matchers 组合在一起。这就是为什么函数式编程最流行的原因，因为它可以重用任何比类级重用更小的函数代码。并且可以看到 jdk 在 jdk-9 中引入了Collectors#filtering，这将使执行路径更短而不会失去其表现力。

AND你可以根据Separation of Concerns进一步重构上面的代码，然后filtering可以像jdk-9一样重用Collectors#filtering：

static <T, U extends T, A, R> Collector<T, ?, R> instanceOf(Class<U> type
        , Collector<U, A, R> downstream) {
  return filtering​(type::isInstance, Collectors.mapping(type::cast, downstream));
}

static <T, A, R>
Collector<T, ?, R> filtering​(Predicate<? super T> predicate
        , Collector<T, A, R> downstream) {
    return new Collector<T, A, R>() {
        @Override
        public Supplier<A> supplier() {
            return downstream.supplier();
        }

        @Override
        public BiConsumer<A, T> accumulator() {
            BiConsumer<A, T> target = downstream.accumulator();
            return (result, it) -> {
                if (predicate.test(it)) {
                    target.accept(result, it);
                }
            };
        }

        @Override
        public BinaryOperator<A> combiner() {
            return downstream.combiner();
        }

        @Override
        public Function<A, R> finisher() {
            return downstream.finisher();
        }

        @Override
        public Set<Characteristics> characteristics() {
            return downstream.characteristics();
        }
    };
}