玖叶教程网

假设我们有以下代码：

List<Car> cars = Arrays.asList(
  new Car("Dacia", "diesel", 100),
  new Car("Lexus", "gasoline", 300),
  ...
  new Car("Ford", "electric", 200)
);
      
Car car1 = new Car("Lexus", "diesel", 300);
Car car2 = new Car("Ford", "electric", 80);
Car car3 = new Car("Chevrolet", "electric", 150);
List<Car> cars123 = List.of(car1, car2, car3);

接下来，在流管道的上下文中，我们想要检查 cars 是否包含 car1、car2、car3 或 cars123 的全部/任意项。Stream API 提供了丰富的中间和最终操作，但它没有内置的 containsAll()/containsAny()。因此，我们的任务是提供以下最终操作：

boolean contains(T item);
boolean containsAll(T... items);
boolean containsAll(List<? extends T> items);
boolean containsAll(Stream<? extends T> items);
boolean containsAny(T... items);
boolean containsAny(List<? extends T> items);
boolean containsAny(Stream<? extends T> items);

我们突出显示了获取 Stream 参数的方法，因为这些方法提供了主要逻辑，而其他方法只是在将参数转换为 Stream 后调用这些方法。

通过自定义接口公开 containsAll/Any() containsAll(Stream<? extends T> items) 依赖于一个 Set 来完成其工作，如下所示（你可以挑战自己找到一个替代实现）：

default boolean containsAll(Stream<? extends T> items) {
  Set<? extends T> set = toSet(items);
  if (set.isEmpty()) {
    return true;
  }
  return stream().filter(item -> set.remove(item))
                 .anyMatch(any -> set.isEmpty());
}

containsAny(Stream<? extends T> items) 也依赖于一个 Set：

default boolean containsAny(Stream<? extends T> items) {
  Set<? extends T> set = toSet(items);
  if (set.isEmpty()) {
    return false;
  }
  return stream().anyMatch(set::contains);
}

toSet() 方法只是一个助手，它将 Stream 项目收集到一个 Set 中：

static <T> Set<T> toSet(Stream<? extends T> stream) {
  return stream.collect(Collectors.toSet());
}

接下来，让我们将这些代码潜入到它们最终的位置，这是一个自定义接口。正如你所看到的，containsAll(Stream<? extends T> items) 和 containsAny(Stream<? extends T> items) 被声明为默认的，这意味着它们是接口的一部分。此外，它们都调用了 stream() 方法，这也是这个接口的一部分，并且连接了常规的 Stream。基本上，解决这个问题的快速方法（在面试中特别有用）是编写这个自定义接口（让我们任意地称它为 Streams），它有权访问原始的内置 Stream 接口，如下所示：

@SuppressWarnings("unchecked")
public interface Streams<T> {
  Stream<T> stream();
  static <T> Streams<T> from(Stream<T> stream) {
    return () -> stream;
  }
  ...
}

接下来，接口公开了一组默认方法，代表 containsAll()/containsAny() 的变体，如下所示：

default boolean contains(T item) {
  return stream().anyMatch(isEqual(item));
}
default boolean containsAll(T... items) {
  return containsAll(Stream.of(items));
}
default boolean containsAll(List<? extends T> items) {
  return containsAll(items.stream());
}
default boolean containsAll(Stream<? extends T> items) {
    ...  
}
default boolean containsAny(T... items) {
  return containsAny(Stream.of(items));
}
default boolean containsAny(List<? extends T> items) {
  return containsAny(items.stream());
}
default boolean containsAny(Stream<? extends T> items) {
    ...
}
static <T> Set<T> toSet(Stream<? extends T> stream) {
    ...
}

完成了！现在，我们可以编写不同的流管道，使用全新的 containsAll/Any() 操作。例如，如果我们想要检查 cars 是否包含 cars123 中的所有项，我们可以这样表达流管道：

boolean result = Streams.from(cars.stream())
  .containsAll(cars123);

这里有更多的例子：

boolean result = Streams.from(cars.stream())
  .containsAll(car1, car2, car3);
boolean result = Streams.from(cars.stream())
  .containsAny(car1, car2, car3);

可以像以下示例那样涉及更多操作：

Car car4 = new Car("Mercedes", "electric", 200);       
boolean result = Streams.from(cars.stream()
    .filter(car->car.getBrand().equals("Mercedes"))
    .distinct()
    .dropWhile(car -> car.getFuel().equals("gasoline"))
  ).contains(car4);

一个更有表现力和完整的解决方案，将包括扩展 Stream 接口。让我们这样做！

通过扩展 Stream 公开 containsAll/Any() 之前的解决方案可以被认为是更像是一种黑客行为。一个更逻辑和现实的解决方案将包括扩展内置的 Stream API，并将我们的 containsAll/Any() 方法作为队友添加到 Stream 操作旁边。因此，实现开始如下：

@SuppressWarnings("unchecked")
public interface Streams<T> extends Stream<T> {       
  ...
}

在实现 containsAll/Any() 方法之前，我们需要处理一些由于扩展 Stream 接口而产生的方面。首先，我们需要在 Streams 中覆盖 Stream 的每个方法。由于 Stream 接口有很多方法，我们这里只列出其中的几种：

@Override
public Streams<T> filter(Predicate<? super T> predicate);
@Override
public <R> Streams<R> map(
  Function<? super T, ? extends R> mapper);
...
@Override
public T reduce(T identity, BinaryOperator<T> accumulator);
...
@Override
default boolean isParallel() {
  return false;
}
...
@Override
default Streams<T> parallel() {
  throw new UnsupportedOperationException(
    "Not supported yet."); // or, return this
}
@Override
default Streams<T> unordered() {
  throw new UnsupportedOperationException(
    "Not supported yet."); // or, return this
}
...
@Override
default Streams<T> sequential() {
  return this;
}

由于 Streams 只能处理顺序流（不支持并行），我们可以在 Streams 中直接实现 isParallel()、parallel()、unordered() 和 sequential() 方法作为默认方法。

接下来，为了使用我们的 Streams，我们需要一个 from(Stream s) 方法，它能够将给定的 Stream 包装如下：

static <T> Streams<T> from(Stream<? extends T> stream) {
  if (stream == null) {
    return from(Stream.empty());
  }
  if (stream instanceof Streams) {
    return (Streams<T>) stream;
  }
  return new StreamsWrapper<>(stream);
}

StreamsWrapper 是一个类，它将当前 Stream 包装成顺序 Streams。StreamsWrapper 实现了 Streams，所以它必须覆盖所有的 Streams 方法，并正确地将 Stream 包装成 Streams。因为 Streams 有很多方法（由于扩展了 Stream 的结果），我们这里只列出其中的几种（其他的在打包的代码中可用）：

@SuppressWarnings("unchecked")
public class StreamsWrapper<T> implements Streams<T> {
  private final Stream<? extends T> delegator;
  public StreamsWrapper(Stream<? extends T> delegator) {
    this.delegator = delegator.sequential();
  }      
  @Override
  public Streams<T> filter(Predicate<? super T> predicate) {       
    return Streams.from(delegator.filter(predicate));
  }
  @Override
  public <R> Streams<R> map(
      Function<? super T, ? extends R> mapper) {
    return Streams.from(delegator.map(mapper));
  }
  ...
  @Override
  public T reduce(T identity, BinaryOperator<T> accumulator) {
    return ((Stream<T>) delegator)
      .reduce(identity, accumulator);
  }
  ...
}

最后，在 Streams 中添加 containsAll/Any() 方法，这些方法非常简单（由于 Streams 扩展了 Stream，我们无需编写 stream() 黑客行为就可以访问所有的 Stream 功能，就像在之前的解决方案中那样）。首先我们添加 containsAll() 方法：

default boolean contains(T item) {
  return anyMatch(isEqual(item));
}
default boolean containsAll(T... items) {
  return containsAll(Stream.of(items));
}
default boolean containsAll(List<? extends T> items) {
  return containsAll(items.stream());
}
default boolean containsAll(Stream<? extends T> items) {
  Set<? extends T> set = toSet(items);
  if (set.isEmpty()) {
    return true;
  }
  return filter(item -> set.remove(item))
    .anyMatch(any -> set.isEmpty());
}

其次，我们添加 containsAny() 方法：

default boolean containsAny(T... items) {
  return containsAny(Stream.of(items));
}
default boolean containsAny(List<? extends T> items) {
  return containsAny(items.stream());
}
default boolean containsAny(Stream<? extends T> items) {
  Set<? extends T> set = toSet(items);
  if (set.isEmpty()) {
    return false;
  }
  return anyMatch(set::contains);
}

以及你已经知道的 toSet() 方法：

static <T> Set<T> toSet(Stream<? extends T> stream) {
  return stream.collect(Collectors.toSet());
}

任务完成！现在，让我们写一些例子：

boolean result = Streams.from(cars.stream())
  .filter(car -> car.getBrand().equals("Mercedes"))
  .contains(car1);
boolean result = Streams.from(cars.stream())
  .containsAll(cars123);
boolean result = Streams.from(cars123.stream())
  .containsAny(cars.stream());