使用窗口

//create a list of 1000 numbers and prepare a Stream to read it
Stream<Integer> sensorDataStream = Streams.from(createTestDataset(1000));

//wait for all windows of 100 to finish
CountDownLatch endLatch = new CountDownLatch(1000 / 100);

Control controls = sensorDataStream
  .window(100)
  .consume(window -> {
    System.out.println("New window starting");
    window
      .reduce(Integer.MAX_VALUE, (acc, next) -> Math.min(acc, next))
      .finallyDo(o -> endLatch.countDown())
      .consume(i -> System.out.println("Minimum " + i));
  });

endLatch.await(10, TimeUnit.SECONDS);
System.out.println(controls.debug());

Assert.assertEquals(0, endLatch.getCount());
Stream<T> API 作用