```java public class SlidingWindowAll { private static final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); private static final DataStreamSource<String> stream = env.socketTextStream("192.168.8.111", 8888); public static void main(String[] args) throws Exception { SingleOutputStreamOperator<Integer> mapped = stream.map((MapFunction<String, Integer>) Integer::valueOf).returns(Types.INT); AllWindowedStream<Integer, TimeWindow> timeWindowAll = mapped.timeWindowAll(Time.seconds(5), Time.seconds(1)); SingleOutputStreamOperator<Integer> summed = timeWindowAll.sum(0); summed.print(); env.execute("SlidingWindowAll"); } } ``` `mapped.timeWindowAll(Time.seconds(5), Time.seconds(1));` 每隔一秒，計算出五秒內的數據。 ```java public class SlidingWindow { private static final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); private static final DataStreamSource<String> stream = env.socketTextStream("192.168.8.111", 8888); public static void main(String[] args) throws Exception { SingleOutputStreamOperator<Tuple2> mapped = stream.map((MapFunction<String, Tuple2>) item -> { String[] data = item.split(" "); return Tuple2.of(data[0], Integer.valueOf(data[1])); }).returns(Types.TUPLE(Types.STRING, Types.INT)); KeyedStream<Tuple2, Tuple> keyed = mapped.keyBy(0); WindowedStream<Tuple2, Tuple, TimeWindow> timeWindow = keyed.timeWindow(Time.seconds(5), Time.seconds(1)); SingleOutputStreamOperator<Tuple2> summed = timeWindow.sum(1); summed.print(); env.execute("SlidingWindow"); } } ``` 這是分組的情況。 timeWindow方法，傳遞一個參數，是滾動窗口。傳入兩個窗口，是滑動窗口。 也可以用Window方法： ```java public class SlidingProcessingTimeWindowsTest { private static final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); private static final DataStreamSource<String> stream = env.socketTextStream("192.168.8.111", 8888); public static void main(String[] args) throws Exception { SingleOutputStreamOperator<Tuple2> mapped = stream.map((MapFunction<String, Tuple2>) item -> { String[] data = item.split(" "); return Tuple2.of(data[0], Integer.valueOf(data[1])); }).returns(Types.TUPLE(Types.STRING, Types.INT)); KeyedStream<Tuple2, Tuple> keyed = mapped.keyBy(0); WindowedStream<Tuple2, Tuple, TimeWindow> timeWindow = keyed.window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1))); SingleOutputStreamOperator<Tuple2> summed = timeWindow.sum(1); summed.print(); env.execute("SlidingProcessingTimeWindowsTest"); } } ```