java - 从流中收集连续对

Question

{ 0, 1, 2, 3, 4 }给定一个流，例如

我怎样才能最优雅地将它转换成给定的形式：

{ new Pair(0, 1), new Pair(1, 2), new Pair(2, 3), new Pair(3, 4) }

（当然，假设我已经定义了类 Pair）？

编辑：这不是严格意义上的整数或原始流。对于任何类型的流，答案都应该是通用的。

score 79 · Accepted Answer

Java 8 流库主要用于将流拆分为较小的块以进行并行处理，因此有状态的管道阶段非常有限，并且不支持获取当前流元素的索引和访问相邻流元素等操作。

当然，解决这些问题的典型方法是通过索引驱动流，并依赖于在一些随机访问数据结构（如 ArrayList）中处理值，从中可以检索元素。如果值在中arrayList，则可以通过执行以下操作生成所请求的对：

    IntStream.range(1, arrayList.size())
             .mapToObj(i -> new Pair(arrayList.get(i-1), arrayList.get(i)))
             .forEach(System.out::println);

当然，限制是输入不能是无限流。不过，这个管道可以并行运行。

score 37 · Accepted Answer

扩展标准流的我的StreamEx库pairMap为所有流类型提供了一种方法。对于原始流，它不会更改流类型，但可用于进行一些计算。最常见的用法是计算差异：

int[] pairwiseDiffs = IntStreamEx.of(input).pairMap((a, b) -> (b-a)).toArray();

对于对象流，您可以创建任何其他对象类型。我的库不提供任何新的用户可见数据结构，例如Pair（这是库概念的一部分）。但是，如果您有自己的Pair课程并想使用它，您可以执行以下操作：

Stream<Pair> pairs = IntStreamEx.of(input).boxed().pairMap(Pair::new);

或者如果你已经有一些Stream：

Stream<Pair> pairs = StreamEx.of(stream).pairMap(Pair::new);

此功能是使用自定义 spliterator实现的。它的开销非常低，并且可以很好地并行化。当然，它适用于任何流源，而不仅仅是像许多其他解决方案那样的随机访问列表/数组。在许多测试中，它表现得非常好。这是一个 JMH 基准测试，我们使用不同的方法找到一个较大值之前的所有输入值（请参阅此问题）。

score 17 · Accepted Answer

这并不优雅，它是一个 hackish 解决方案，但适用于无限流

Stream<Pair> pairStream = Stream.iterate(0, (i) -> i + 1).map( // natural numbers
    new Function<Integer, Pair>() {
        Integer previous;

        @Override
        public Pair apply(Integer integer) {
            Pair pair = null;
            if (previous != null) pair = new Pair(previous, integer);
            previous = integer;
            return pair;
        }
    }).skip(1); // drop first null

现在您可以将流限制为您想要的长度

pairStream.limit(1_000_000).forEach(i -> System.out.println(i));

PS我希望有更好的解决方案，比如clojure(partition 2 1 stream)

score 17 · Accepted Answer

您可以使用 Stream.reduce() 方法来做到这一点（我还没有看到使用这种技术的任何其他答案）。

public static <T> List<Pair<T, T>> consecutive(List<T> list) {
    List<Pair<T, T>> pairs = new LinkedList<>();
    list.stream().reduce((a, b) -> {
        pairs.add(new Pair<>(a, b));
        return b;
    });
    return pairs;
}

score 16 · Accepted Answer

我已经实现了一个拆分器包装器，它从原始拆分器中获取每个n元素并生成：TList<T>

public class ConsecutiveSpliterator<T> implements Spliterator<List<T>> {

    private final Spliterator<T> wrappedSpliterator;

    private final int n;

    private final Deque<T> deque;

    private final Consumer<T> dequeConsumer;

    public ConsecutiveSpliterator(Spliterator<T> wrappedSpliterator, int n) {
        this.wrappedSpliterator = wrappedSpliterator;
        this.n = n;
        this.deque = new ArrayDeque<>();
        this.dequeConsumer = deque::addLast;
    }

    @Override
    public boolean tryAdvance(Consumer<? super List<T>> action) {
        deque.pollFirst();
        fillDeque();
        if (deque.size() == n) {
            List<T> list = new ArrayList<>(deque);
            action.accept(list);
            return true;
        } else {
            return false;
        }
    }

    private void fillDeque() {
        while (deque.size() < n && wrappedSpliterator.tryAdvance(dequeConsumer))
            ;
    }

    @Override
    public Spliterator<List<T>> trySplit() {
        return null;
    }

    @Override
    public long estimateSize() {
        return wrappedSpliterator.estimateSize();
    }

    @Override
    public int characteristics() {
        return wrappedSpliterator.characteristics();
    }
}

以下方法可用于创建连续流：

public <E> Stream<List<E>> consecutiveStream(Stream<E> stream, int n) {
    Spliterator<E> spliterator = stream.spliterator();
    Spliterator<List<E>> wrapper = new ConsecutiveSpliterator<>(spliterator, n);
    return StreamSupport.stream(wrapper, false);
}

示例用法：

consecutiveStream(Stream.of(0, 1, 2, 3, 4, 5), 2)
    .map(list -> new Pair(list.get(0), list.get(1)))
    .forEach(System.out::println);

score 7 · Accepted Answer

你可以在cyclops-react（我贡献了这个库）中使用滑动操作符来做到这一点。

  LazyFutureStream.of( 0, 1, 2, 3, 4 )
                  .sliding(2)
                  .map(Pair::new);

或者

   ReactiveSeq.of( 0, 1, 2, 3, 4 )
                  .sliding(2)
                  .map(Pair::new);

假设 Pair 构造函数可以接受具有 2 个元素的 Collection。

如果您想按 4 分组，并按 2 递增，这也是受支持的。

     ReactiveSeq.rangeLong( 0L,Long.MAX_VALUE)
                .sliding(4,2)
                .forEach(System.out::println);

在 cyclops-streams StreamUtils类中还提供了用于在 java.util.stream.Stream 上创建滑动视图的等效静态方法。

       StreamUtils.sliding(Stream.of(1,2,3,4),2)
                  .map(Pair::new);

注意：- 对于单线程操作 ReactiveSeq 会更合适。LazyFutureStream 扩展了 ReactiveSeq，但主要面向并发/并行使用（它是期货流）。

LazyFutureStream 扩展了 ReactiveSeq，它从令人敬畏的 jOOλ（扩展了 java.util.stream.Stream）扩展了 Seq，因此 Lukas 提出的解决方案也适用于任一 Stream 类型。对于任何感兴趣的人来说，窗口/滑动操作符之间的主要区别是明显的相对功率/复杂性权衡以及与无限流一起使用的适用性（滑动不消耗流，而是在流动时缓冲）。

score 6 · Accepted Answer

Streams.zip(..)在 Guava 中可用，供依赖它的人使用。

例子：

Streams.zip(list.stream(),
            list.stream().skip(1),
            (a, b) -> System.out.printf("%s %s\n", a, b));

score 5 · Accepted Answer

寻找连续的对

如果您愿意使用第三方库并且不需要并行性，那么jOOλ提供如下 SQL 风格的窗口函数

System.out.println(
Seq.of(0, 1, 2, 3, 4)
   .window()
   .filter(w -> w.lead().isPresent())
   .map(w -> tuple(w.value(), w.lead().get())) // alternatively, use your new Pair() class
   .toList()
);

屈服

[(0, 1), (1, 2), (2, 3), (3, 4)]

该lead()函数以遍历顺序从窗口访问下一个值。

寻找连续的三元组/四元组/n元组

评论中的一个问题是要求一个更通用的解决方案，其中不应该收集对，而是应该收集 n 元组（或可能是列表）。因此，这是一种替代方法：

int n = 3;

System.out.println(
Seq.of(0, 1, 2, 3, 4)
   .window(0, n - 1)
   .filter(w -> w.count() == n)
   .map(w -> w.window().toList())
   .toList()
);

生成列表列表

[[0, 1, 2], [1, 2, 3], [2, 3, 4]]

没有filter(w -> w.count() == n)，结果将是

[[0, 1, 2], [1, 2, 3], [2, 3, 4], [3, 4], [4]]

免责声明：我为 jOOλ 背后的公司工作

score 4 · Accepted Answer

proton-pack 库提供了窗口化的功能。给定一个 Pair 类和一个 Stream，你可以这样做：

Stream<Integer> st = Stream.iterate(0 , x -> x + 1);
Stream<Pair<Integer, Integer>> pairs = StreamUtils.windowed(st, 2, 1)
                                                  .map(l -> new Pair<>(l.get(0), l.get(1)))
                                                  .moreStreamOps(...);

现在pairs流包含：

(0, 1)
(1, 2)
(2, 3)
(3, 4)
(4, ...) and so on

score 2 · Accepted Answer

该操作本质上是有状态的，因此并不是真正要解决的流 - 请参阅javadoc中的“无状态行为”部分：

最好的方法是避免有状态的行为参数完全流式操作

这里的一种解决方案是通过外部计数器在流中引入状态，尽管它仅适用于顺序流。

public static void main(String[] args) {
    Stream<String> strings = Stream.of("a", "b", "c", "c");
    AtomicReference<String> previous = new AtomicReference<>();
    List<Pair> collect = strings.map(n -> {
                            String p = previous.getAndSet(n);
                            return p == null ? null : new Pair(p, n);
                        })
                        .filter(p -> p != null)
                        .collect(toList());
    System.out.println(collect);
}


static class Pair<T> {
    private T left, right;
    Pair(T left, T right) { this.left = left; this.right = right; }
    @Override public String toString() { return "{" + left + "," + right + '}'; }
}

score 2 · Accepted Answer

我们可以使用RxJava（非常强大的响应式扩展库）

IntStream intStream  = IntStream.iterate(1, n -> n + 1);

Observable<List<Integer>> pairObservable = Observable.from(intStream::iterator).buffer(2,1);

pairObservable.take(10).forEach(b -> {
            b.forEach(n -> System.out.println(n));
            System.out.println();
        });

缓冲区 操作符将发出项目的 Observable 转换为发出这些项目的缓冲集合的 Observable。

score 0 · Accepted Answer

在你的情况下，我会编写我的自定义 IntFunction 来跟踪最后一个传递的 int 并使用它来映射原始 IntStream。

import java.util.function.IntFunction;
import java.util.stream.IntStream;

public class PairFunction implements IntFunction<PairFunction.Pair> {

  public static class Pair {

    private final int first;
    private final int second;

    public Pair(int first, int second) {
      this.first = first;
      this.second = second;
    }

    @Override
    public String toString() {
      return "[" + first + "|" + second + "]";
    }
  }

  private int last;
  private boolean first = true;

  @Override
  public Pair apply(int value) {
    Pair pair = !first ? new Pair(last, value) : null;
    last = value;
    first = false;
    return pair;
  }

  public static void main(String[] args) {

    IntStream intStream = IntStream.of(0, 1, 2, 3, 4);
    final PairFunction pairFunction = new PairFunction();
    intStream.mapToObj(pairFunction)
        .filter(p -> p != null) // filter out the null
        .forEach(System.out::println); // display each Pair

  }

}

score 0 · Accepted Answer

一个优雅的解决方案是使用zip。就像是：

List<Integer> input = Arrays.asList(0, 1, 2, 3, 4);
Stream<Pair> pairStream = Streams.zip(input.stream(),
                                      input.stream().substream(1),
                                      (a, b) -> new Pair(a, b)
);

这非常简洁和优雅，但是它使用列表作为输入。无法以这种方式处理无限流源。

另一个（更麻烦的）问题是 zip 连同整个 Streams 类最近已从 API 中删除。上面的代码只适用于b95或更早的版本。因此，对于最新的 JDK，我想说没有优雅的 FP 风格的解决方案，现在我们只能希望以某种方式将 zip 重新引入 API。

score 0 · Accepted Answer

我终于找到了一种欺骗 Stream.reduce 的方法，以便能够巧妙地处理成对的值；有许多用例需要此功能，而这在 JDK 8中并不自然出现：

public static int ArithGeo(int[] arr) {
    //Geometric
    List<Integer> diffList = new ArrayList<>();
    List<Integer> divList = new ArrayList<>();
    Arrays.stream(arr).reduce((left, right) -> {
        diffList.add(right-left);
        divList.add(right/left);
        return right;
    });
    //Arithmetic
    if(diffList.stream().distinct().count() == 1) {
        return 1;
    }
    //Geometric
    if(divList.stream().distinct().count() == 1) {
        return 2;
    }
    return -1;
}

我使用的技巧是返回权；陈述。

score 0 · Accepted Answer

为了计算时间序列的时间（x 值）的连续差异，我使用stream'collect(...)方法：

final List< Long > intervals = timeSeries.data().stream()
                    .map( TimeSeries.Datum::x )
                    .collect( DifferenceCollector::new, DifferenceCollector::accept, DifferenceCollector::combine )
                    .intervals();

DifferenceCollector 是这样的：

public class DifferenceCollector implements LongConsumer
{
    private final List< Long > intervals = new ArrayList<>();
    private Long lastTime;

    @Override
    public void accept( final long time )
    {
        if( Objects.isNull( lastTime ) )
        {
            lastTime = time;
        }
        else
        {
            intervals.add( time - lastTime );
            lastTime = time;
        }
    }

    public void combine( final DifferenceCollector other )
    {
        intervals.addAll( other.intervals );
        lastTime = other.lastTime;
    }

    public List< Long > intervals()
    {
        return intervals;
    }
}

您可能可以修改它以满足您的需要。

score -1 · Accepted Answer

正如其他人所观察到的，由于问题的性质，需要一些状态性。

我遇到了类似的问题，我想要的是本质上是 Oracle SQL 函数 LEAD。我的尝试如下。

/**
 * Stream that pairs each element in the stream with the next subsequent element.
 * The final pair will have only the first item, the second will be null.
 */
<T> Spliterator<Pair<T>> lead(final Stream<T> stream)
{
    final Iterator<T> input = stream.sequential().iterator();

    final Iterable<Pair<T>> iterable = () ->
    {
        return new Iterator<Pair<T>>()
        {
            Optional<T> current = getOptionalNext(input);

            @Override
            public boolean hasNext()
            {
                return current.isPresent();
            }

            @Override
            public Pair<T> next()
            {
                Optional<T> next = getOptionalNext(input);
                final Pair<T> pair = next.isPresent()
                    ? new Pair(current.get(), next.get())
                    : new Pair(current.get(), null);
                current = next;

                return pair;
            }
        };
    };

    return iterable.spliterator();
}

private <T> Optional<T> getOptionalNext(final Iterator<T> iterator)
{
    return iterator.hasNext()
        ? Optional.of(iterator.next())
        : Optional.empty();
}

score -1 · Accepted Answer

这是一个有趣的问题。我的混合尝试有什么好处吗？

public static void main(String[] args) {
    List<Integer> list = Arrays.asList(1, 2, 3);
    Iterator<Integer> first = list.iterator();
    first.next();
    if (first.hasNext())
        list.stream()
        .skip(1)
        .map(v -> new Pair(first.next(), v))
        .forEach(System.out::println);
}

我相信它不适合并行处理，因此可能会被取消资格。

score -1 · Accepted Answer

您可以通过使用有界队列来存储流经流的元素来实现这一点（这是基于我在这里详细描述的想法：是否可以在流中获取下一个元素？）

下面的示例首先定义了 BoundedQueue 类的实例，它将存储通过流的元素（如果您不喜欢扩展 LinkedList 的想法，请参阅上面提到的链接以获取替代和更通用的方法）。稍后您只需将两个后续元素组合到 Pair 的实例中：

public class TwoSubsequentElems {
  public static void main(String[] args) {
    List<Integer> input = new ArrayList<Integer>(asList(0, 1, 2, 3, 4));

    class BoundedQueue<T> extends LinkedList<T> {
      public BoundedQueue<T> save(T curElem) {
        if (size() == 2) { // we need to know only two subsequent elements
          pollLast(); // remove last to keep only requested number of elements
        }

        offerFirst(curElem);

        return this;
      }

      public T getPrevious() {
        return (size() < 2) ? null : getLast();
      }

      public T getCurrent() {
        return (size() == 0) ? null : getFirst();
      }
    }

    BoundedQueue<Integer> streamHistory = new BoundedQueue<Integer>();

    final List<Pair<Integer>> answer = input.stream()
      .map(i -> streamHistory.save(i))
      .filter(e -> e.getPrevious() != null)
      .map(e -> new Pair<Integer>(e.getPrevious(), e.getCurrent()))
      .collect(Collectors.toList());

    answer.forEach(System.out::println);
  }
}

score -3 · Accepted Answer

我同意@aepurniet 但你必须使用 mapToObj 来代替 map

range(0, 100).mapToObj((i) -> new Pair(i, i+1)).forEach(System.out::println);

score -6 · Accepted Answer

运行for从 0 到length-1流的循环

for(int i = 0 ; i < stream.length-1 ; i++)
{
    Pair pair = new Pair(stream[i], stream[i+1]);
    // then add your pair to an array
}

java - 从流中收集连续对

20 回答 20

寻找连续的对

寻找连续的三元组/四元组/n元组

Related

Reference