Java 8 一个流到多个映射

Question

假设我有一个巨大的网络服务器日志文件，内存无法容纳。我需要将此文件流式传输到 mapreduce 方法并保存到数据库。我使用 Java 8 stream api 执行此操作。例如，我在 mapreduce 过程之后得到一个列表，如客户端消费、ip 消费、内容消费。但是，我的需求并不像我的示例中给出的那样。由于我不能分享代码，我只想举一个基本的例子。

通过 Java 8 Stream Api，我想读取文件一次，同时获取 3 个列表，同时我正在流式传输文件，并行或顺序。但是并行会很好。有什么办法吗？

Answer 1

通常，通过自定义 Collector 收集标准 API 以外的任何内容都非常容易。在您的情况下，一次收集到 3 个列表(只是一个编译的小示例，因为您也不能共享您的代码):

private static <T> Collector<T, ?, List<List<T>>> to3Lists() {
    class Acc {

        List<T> left = new ArrayList<>();

        List<T> middle = new ArrayList<>();

        List<T> right = new ArrayList<>();

        List<List<T>> list = Arrays.asList(left, middle, right);

        void add(T elem) {
            // obviously do whatever you want here
            left.add(elem);
            middle.add(elem);
            right.add(elem);
        }

        Acc merge(Acc other) {

            left.addAll(other.left);
            middle.addAll(other.middle);
            right.addAll(other.right);

            return this;
        }

        public List<List<T>> finisher() {
            return list;
        }

    }
    return Collector.of(Acc::new, Acc::add, Acc::merge, Acc::finisher);
}

并通过以下方式使用它:

Stream.of(1, 2, 3)
      .collect(to3Lists());

显然，这个自定义收集器没有做任何有用的事情，只是一个如何使用它的示例。