StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.execute("My Flink Job");

public static void main(String[] args) throws Exception {
    // 1. 获取 Flink 流处理执行环境
    // 这是所有 Flink 程序的入口点，用于构建和提交数据流作业
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // 2. 创建一个 DataStream（数据流），使用 fromElements() 方法从一组预定义的 Person 对象构造源流
    // 此处模拟了一个小型静态数据集，包含三个人的信息
    DataStream<Person> flintstones = env.fromElements(
            new Person("Fred", 35),     // 成年人
            new Person("Wilma", 35),    // 成年人
            new Person("Pebbles", 2)    // 小孩
    );

    // 3. 对数据流进行转换：使用 filter() 算子过滤出年龄大于等于 18 岁的 Person 对象
    // filter 接收一个 Predicate（布尔函数），返回 true 的元素会被保留在结果流中
    DataStream<Person> adults = flintstones.filter(person -> person.age >= 18);

    // 4. 添加 Sink：将过滤后的结果打印到标准输出（控制台）
    // print() 是一个内置的 sink 算子，常用于调试和演示
    // 输出格式为 "<subtask_id> <element>"，例如 "1> Fred: age 35"
    adults.print();

    // 5. 触发作业执行
    // 至此，Flink 才会真正构建并提交数据流图（DAG）到本地或集群运行
    // execute() 方法会阻塞主线程，直到作业完成（对于有界流）或被取消（对于无界流）
    env.execute();
}

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class BoundedJoinExample {

    public static void main(String[] args) throws Exception {
        // 1. 创建环境，并显式设置为 BATCH 模式（关键！）
        StreamExecution环境 env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH); // ← 启用批优化

        // 2. 模拟两个有界数据源（实际可用 readTextFile / fromCollection）
        DataStream<Tuple3<String, String, Double>> orders = env.fromCollection(java.util.Arrays.asList(
            Tuple3.of("order1", "user1", 100.0),
            Tuple3.of("order2", "user2", 200.0),
            Tuple3.of("order3", "user1", 150.0),
            Tuple3.of("order4", "user3", 300.0)
        ));

        DataStream<Tuple2<String, String>> users = env.fromCollection(java.util.Arrays.asList(
            Tuple2.of("user1", "beijing"),
            Tuple2.of("user2", "shanghai"),
            Tuple2.of("user3", "guangzhou")
        ));

        // 3. 给数据加上“事件时间”（这里用固定时间，因为是有界流）
        WatermarkStrategy<Tuple3<String, String, Double>> orderWm = WatermarkStrategy.noWatermarks();
        WatermarkStrategy<Tuple2<String, String>> userWm = WatermarkStrategy.noWatermarks();

        DataStream<Tuple3<String, String, Double>> ordersWithTs = orders.assignTimestampsAndWatermarks(orderWm);
        DataStream<Tuple2<String, String>> usersWithTs = users.assignTimestampsAndWatermarks(userWm);

        // 4. Join：按 userId 关联订单和用户
        DataStream<Tuple3<String, String, Double>> enriched = ordersWithTs
            .keyBy(order -> order.f1) // userId
            .intervalJoin(usersWithTs.keyBy(user -> user.f0))
            .between(Time.milliseconds(-1), Time.milliseconds(1)) // 因为是静态数据，时间窗口设为0
            .process(new JoinFunction<...>() {
                @Override
                public Tuple3<String, String, Double> join(Tuple3<String, String, Double> order, Tuple2<String, String> user) {
                    return Tuple3.of(user.f1, order.f1, order.f2); // (region, userId, amount)
                }
            });

        // 5. 按 region 分组，统计总金额（这里用滚动窗口，但因为是有界流，窗口会一次性触发）
        enriched
            .keyBy(record -> record.f0) // region
            .window(TumblingEventTimeWindows.of(Time.days(1))) // 假设所有数据在同一天
            .reduce((a, b) -> Tuple3.of(a.f0, "SUM", a.f2 + b.f2)) // 累加金额
            .print();

        // 6. 执行
        env.execute("Bounded Join Example");
    }
}

DataStream<String> lines = env.fromElements("hello world", "flink stream");
DataStream<String> words = lines.flatMap((line, out) -> {
    for (String word : line.split(" ")) {
        out.collect(word); // 可输出多个
    }
});
// 输出: hello, world, flink, stream

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2); // 全局并行度 = 2

// 创建一个包含 4 个元素的有界流
DataStream<String> source = env.fromElements("A", "B", "C", "D");

// map 算子（未显式设置，并行度=2）
DataStream<Integer> mapped = source.map(s -> {
    System.out.println("Thread: " + Thread.currentThread().getName() + " processing: " + s);
    return s.length();
});

mapped.print(); // sink 并行度也=2
env.execute();

DataStream<Person> people = env.fromElements(
    new Person("Alice", 25),
    new Person("Bob", 30),
    new Person("Alice", 28),
    new Person("Charlie", 22)
);

// 假设并行度 = 2
KeyedStream<Person, String> keyed = people.keyBy(p -> p.name);

class Order {
    String orderId;
    long eventTime; // 订单实际发生的时间（毫秒时间戳）
}

// 允许最多 5 秒的乱序（即：事件最多延迟 5 秒到达）
WatermarkStrategy<Order> strategy = WatermarkStrategy
    .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withTimestampAssigner((event, timestamp) -> event.eventTime);

.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.seconds(10)) // 窗口关闭后，再等 10 秒
.aggregate(...)

OutputTag<Order> lateTag = new OutputTag<Order>("late-orders") {};

SingleOutputStreamOperator<Long> result = stream
    .window(...)
    .sideOutputLateData(lateTag)
    .aggregate(...);

// 单独处理迟到数据
result.getSideOutput(lateTag).addSink(...); // 如写入告警日志

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

public class OrderCountWithEventTime {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // Flink 1.12+ 默认使用 Event Time，但显式设置更清晰
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.enableCheckpointing(5000); // 开启 checkpoint，保障容错

        // 模拟订单数据源
        DataStream<Order> orders = env.addSource(new OrderSourceFunction());

        // 定义 Watermark 策略：允许 5 秒乱序
        WatermarkStrategy<Order> strategy = WatermarkStrategy
            .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((order, ts) -> order.eventTime)
            .withWatermarkInterval(Duration.ofMillis(200)); // 可选：调整生成频率

        DataStream<Order> withWatermarks = orders.assignTimestampsAndWatermarks(strategy);

        // 定义迟到数据的 Side Output
        OutputTag<Order> lateOrdersTag = new OutputTag<>("late-orders"){};

        // 每分钟滚动窗口统计
        withWatermarks
            .keyBy(order -> "global") // 全局统计（不分组）
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .allowedLateness(Time.seconds(10)) // 允许额外 10 秒迟到
            .sideOutputLateData(lateOrdersTag)
            .aggregate(new CountAggregate()) // 自定义聚合函数
            .print("On-time Result");

        // 输出超时迟到事件
        withWatermarks
            .getSideOutput(lateOrdersTag)
            .print("Late Order");

        env.execute("Real-time Order Count per Minute (Event Time)");
    }
}

WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))
// ...
.window(...).allowedLateness(Time.seconds(10))

.withIdleness(Duration.ofSeconds(30)) // 30秒无数据，Watermark 自动推进

public abstract class WindowAssigner<T, W extends Window> implements Serializable {
        /**     
        * 为每一个输入元素分配窗口.     
        *     
        * @param element 输入元素     
        * @param timestamp 元素的时间戳。如果没有通过assignTimestampsAndWatermarks方法分配时间戳，则为Long.MIN_VALUE     
        * @param context 窗口上下文     
        * @return 窗口集合     
        */    
        public abstract Collection<W> assignWindows(T element, long timestamp, 
                                                    WindowAssigner.WindowAssignerContext context);    
        
        /**     
        * 获取窗口的默认触发器.     
        *      
        * @param env 流模式环境     
        * @return 触发器     
        */    
        public abstract Trigger<T, W> getDefaultTrigger(StreamExecutionEnvironment env);    
        
}

// 触发器枚举
public enum TriggerResult {    
  // 不做任何处理   
  CONTINUE(false, false),    

  // 触发窗口计算，并清空窗口内的数据   
  FIRE_AND_PURGE(true, true),    

  // 触发窗口计算，不清空窗口内的数据 
  FIRE(true, false),    

  // 清空窗口内的数据 
  PURGE(false, true);  

  // 是否触发计算
  private final boolean fire;    

  // 是否清空窗口
  private final boolean purge;

}


public abstract class Trigger<T, W extends Window> implements Serializable {        

  // 当接收到一条输入元素时，判断触发器的结果    
  public abstract TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx);          throws Exception;    

  // 到达处理时间Timer注册的时间，会判断触发器的结果    
  public abstract TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception;   

   // 到达事件时间Timer注册的时间，会判断触发器的结果    
   public abstract TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception;

}

public class TimeWindow extends Window {
    private final long start;
    private final long end;

    public long maxTimestamp() {
        return end - 1L; // 窗口内允许的最大事件时间戳
    }
}

// 滚动事件时间窗口：每 5 分钟统计一次
stream
  .keyBy(...)
  .window(TumblingEventTimeWindows.of(Time.minutes(5)))
  .sum("value");

// 滑动事件时间窗口：每 1 分钟统计最近 5 分钟的数据
stream
  .keyBy(...)
  .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
  .sum("value");

stream.countWindow(100); 
// 实际是：window(GlobalWindows.create()).trigger(CountTrigger.of(100))

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;

public class CountWindowExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 模拟订单流：(orderId, amount)
        DataStream<Order> orderStream = env.fromElements(
            new Order("1", 10.0),
            new Order("2", 20.0),
            new Order("3", 30.0),  // 触发第1次计算（3条）
            new Order("4", 5.0),
            new Order("5", 15.0),
            new Order("6", 25.0),  // 触发第2次计算（又3条）
            new Order("7", 40.0)
        );

        // 使用计数窗口：每3条触发一次
        orderStream
            .keyBy(order -> "global") // 如果不需要 keyBy，也可用 allWindow（见下方说明）
            .window(GlobalWindows.create())
            .trigger(CountTrigger.of(3))
            .sum("amount")
            .print(); // 输出：60.0, 45.0, ...

        env.execute("Count Window Example");
    }

    public static class Order {
        public String id;
        public double amount;

        public Order(String id, double amount) {
            this.id = id;
            this.amount = amount;
        }
    }
}

.window(GlobalWindows.create())
.trigger(CountTrigger.of(100))
.process(new ProcessWindowFunction<...>())

stream
  .keyBy(...)
  .window(GlobalWindows.create())
  .trigger(CountTrigger.of(100))
  .reduce((value1, value2) -> new Order(value1.amount + value2.amount))
  .print();

stream
  .keyBy(...)
  .window(GlobalWindows.create())
  .trigger(CountTrigger.of(100))
  .evictor(CountEvictor.of(100)) // 触发前移除最近100条
  .process(new ProcessWindowFunction<...>() { ... });

// 安全：状态恒定，无增长风险
stream
  .keyBy("userId")
  .window(GlobalWindows.create())
  .trigger(CountTrigger.of(1000))
  .reduce(
      (order1, order2) -> new Order(order1.userId, order1.amount + order2.amount),
      (key, window, reduced, out) -> out.collect(reduced)
  )
  .print();

stream
  .keyBy("userId")
  .window(TumblingEventTimeWindows.of(Time.minutes(1)))
  .reduce((order1, order2) -> new Order(order1.amount + order2.amount))
  .print();



-- 等价于
.reduce((ReduceFunction<Order>) (order1, order2) -> 
    new Order(order1.amount + order2.amount)
)

stream
  .keyBy("service")
  .window(TumblingProcessingTimeWindows.of(Time.seconds(30)))
  .process(new ProcessWindowFunction<LatencyEvent, String, String, TimeWindow>() {
      @Override
      public void process(String key, Context ctx, Iterable<LatencyEvent> elements, Collector<String> out) {
          List<Long> latencies = new ArrayList<>();
          for (LatencyEvent e : elements) {
              latencies.add(e.latencyMs);
          }
          Collections.sort(latencies);
          int idx = (int) (0.999 * latencies.size());
          long tp999 = latencies.get(idx);
          out.collect("Service=" + key + ", TP999=" + tp999 + "ms");
      }
  });

// 1. 定义增量聚合：计算 (sum, count)
AggregateFunction<LatencyEvent, Tuple2<Long, Integer>, Double> avgAgg = new AggregateFunction<...>() {
    public Tuple2<Long, Integer> createAccumulator() { return Tuple2.of(0L, 0); }
    public Tuple2<Long, Integer> add(LatencyEvent e, Tuple2<Long, Integer> acc) {
        return Tuple2.of(acc.f0 + e.latencyMs, acc.f1 + 1);
    }
    public Double getResult(Tuple2<Long, Integer> acc) {
        return acc.f1 == 0 ? 0.0 : (double) acc.f0 / acc.f1;
    }
    public Tuple2<Long, Integer> merge(...) { ... }
};

// 2. 定义后处理：加上窗口信息
ProcessWindowFunction<Double, MetricRecord, String, TimeWindow> enricher = new ProcessWindowFunction<...>() {
    public void process(String key, Context ctx, Iterable<Double> averages, Collector<MetricRecord> out) {
        Double avg = averages.iterator().next(); // 只有一个值
        TimeWindow w = ctx.window();
        out.collect(new MetricRecord(key, avg, w.getStart(), w.getEnd()));
    }
};

// 3. 组合使用
stream
  .keyBy("service")
  .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
  .aggregate(avgAgg, enricher); // ← 关键：两个参数！

stream
  .keyBy(...)                 // 1. 按 key 分组（必须！）
  .window(...)                // 2. 定义窗口类型
  .allowedLateness(...)       // 3. （可选）允许迟到数据
  .reduce/process/fold/apply  // 4. 聚合函数

public static class ClickEvent {
    public String userId;
    public long eventTime; // 毫秒时间戳

    public ClickEvent(String userId, long eventTime) {
        this.userId = userId;
        this.eventTime = eventTime;
    }

    @Override
    public String toString() {
        return userId + " clicked at " + eventTime;
    }
}

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(5000);

// 模拟乱序点击流
DataStream<ClickEvent> clicks = env.fromCollection(Arrays.asList(
    new ClickEvent("user1", 1700000000000L), // 2023-11-14 12:26:40
    new ClickEvent("user1", 1700000000005L),
    new ClickEvent("user2", 1700000000010L),
    new ClickEvent("user1", 1700000000002L)  // 迟到事件（早于上一条）
));

// 分配时间戳 + 生成 Watermark（允许5秒乱序）
WatermarkStrategy<ClickEvent> strategy = WatermarkStrategy
    .<ClickEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withTimestampAssigner((event, ts) -> event.eventTime);

DataStream<ClickEvent> withWm = clicks.assignTimestampsAndWatermarks(strategy);

// 每1分钟窗口，统计每个用户的点击次数
DataStream<String> result = withWm
    .keyBy(event -> event.userId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new CountAggregate(), new FormatResult());

result.print();
env.execute("User Click Count per Minute");

// 累加器：只存计数
public static class CountAggregate implements AggregateFunction<ClickEvent, Long, Long> {
    @Override
    public Long createAccumulator() { return 0L; }
    @Override
    public Long add(ClickEvent event, Long acc) { return acc + 1; }
    @Override
    public Long getResult(Long acc) { return acc; }
    @Override
    public Long merge(Long a, Long b) { return a + b; }
}

// 输出格式化
public static class FormatResult implements WindowFunction<Long, String, String, TimeWindow> {
    @Override
    public void apply(String userId, TimeWindow window, Iterable<Long> counts, Collector<String> out) {
        long count = counts.iterator().next();
        String start = new Timestamp(window.getStart()).toString();
        String end = new Timestamp(window.getEnd()).toString();
        out.collect(userId + " clicked " + count + " times in [" + start + ", " + end + ")");
    }
}

OutputTag<ClickEvent> lateTag = new OutputTag<ClickEvent>("late-clicks") {};

SingleOutputStreamOperator<String> mainStream = withWm
    .keyBy(e -> e.userId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .allowedLateness(Time.seconds(10))      // 允许额外等10秒
    .sideOutputLateData(lateTag)            // 超过10秒的走旁路
    .aggregate(new CountAggregate(), new FormatResult());

// 获取迟到流
DataStream<ClickEvent> lateStream = mainStream.getSideOutput(lateTag);
lateStream.print("LATE"); // 输出前缀 "LATE> ..."

mainStream.print("MAIN");

public static class WordCountFunction extends RichMapFunction<String, Tuple2<String, Long>> {
    private ValueState<Long> countState;

    @Override
    public void open(Configuration parameters) {
        // 定义状态描述符：名称 + 类型
        ValueStateDescriptor<Long> descriptor =
            new ValueStateDescriptor<>("word-count", Long.class);
        countState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public Tuple2<String, Long> map(String word) throws Exception {
        Long current = countState.value();
        if (current == null) current = 0L;
        current += 1;
        countState.update(current); // 更新状态
        return Tuple2.of(word, current);
    }
}

// 使用
stream.keyBy(word -> word)
      .map(new WordCountFunction())
      .print();

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(1)) // 过期时间 = 1天
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) // 何时更新TTL？
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) // 过期后是否可见？
    .cleanupInRocksDBCompactFilter(1000) // RocksDB后台清理（可选）
    .build();

public static class ActiveUserFunction extends RichFlatMapFunction<String, String> {
    private ValueState<Boolean> activeState;

    @Override
    public void open(Configuration conf) {
        StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(Time.hours(2))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
            .build();

        ValueStateDescriptor<Boolean> descriptor = new ValueStateDescriptor<>("active-flag", Boolean.class);
        descriptor.enableTimeToLive(ttlConfig);

        activeState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap(String userId, Collector<String> out) {
        activeState.update(true); // 标记用户活跃，并重置TTL
        out.collect(userId + " is active now");
    }
}

stream
    .keyBy(...)
    .window(...)
    .allowedLateness(Time.seconds(10))
    .process(...);

@Override
public void processElement(LoginEvent event, Context ctx, Collector<String> out) {
    long eventTime = ctx.timestamp();          // 事件时间戳
    String userId = ctx.getCurrentKey();        // 当前 key（需 keyBy 后才有）
    TimerService timer = ctx.timerService();   // 获取定时器服务

    // 注册一个 10 分钟后的事件时间定时器
    timer.registerEventTimeTimer(eventTime + 10 * 60 * 1000);
}

@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) {
    String userId = ctx.getCurrentKey();
    out.collect("⏰ Timeout for user: " + userId);

    // 可在此清理状态
    myState.clear();
}

// 事件时间 Timer：基于业务时间（可靠，但可能延迟）
ctx.timerService().registerEventTimeTimer(event.eventTime + 30 * 60 * 1000);

// 处理时间 Timer：基于机器时间（立即，但不精确）
ctx.timerService().registerProcessingTimeTimer(System.currentTimeMillis() + 30_000);

public class SessionTimeoutFunction 
    extends KeyedProcessFunction<String, UserEvent, String> {

    // 状态：记录会话开始时间
    private ValueState<Long> sessionStartTime;

    @Override
    public void open(Configuration parameters) {
        sessionStartTime = getRuntimeContext()
            .getState(new ValueStateDescriptor<>("start-time", Long.class));
    }

    @Override
    public void processElement(UserEvent event, Context ctx, Collector<String> out) throws Exception {
        long currentTime = event.timestamp;

        // 如果是新会话（状态为空 或 距离上次 > 30分钟）
        Long start = sessionStartTime.value();
        if (start == null || currentTime - start > 30 * 60 * 1000) {
            // 开启新会话
            sessionStartTime.update(currentTime);
        }

        // 每次事件都更新：注册一个 30 分钟后的 Timer
        ctx.timerService().registerEventTimeTimer(currentTime + 30 * 60 * 1000);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        Long start = sessionStartTime.value();
        if (start != null) {
            long duration = timestamp - start;
            out.collect(ctx.getCurrentKey() + " session ended, duration: " + duration / 1000 + "s");
            sessionStartTime.clear(); // 清理会话
        }
    }
}

// 伪代码：WindowOperator 的核心逻辑
class WindowOperator extends KeyedProcessFunction<...> {
    private MapState<TimeWindow, Accumulator> windowStates; // 窗口状态

    @Override
    public void processElement(Event e, Context ctx, Collector out) {
        // 1. 将事件分配给多个窗口（Sliding）
        for (TimeWindow w : assignWindows(e)) {
            // 2. 更新该窗口的状态
            Accumulator acc = windowStates.get(w);
            if (acc == null) acc = createAcc();
            acc = add(e, acc);
            windowStates.put(w, acc);

            // 3. 注册窗口结束时间的 Timer
            ctx.timerService().registerEventTimeTimer(w.getEnd());
        }
    }

    @Override
    public void onTimer(long ts, OnTimerContext ctx, Collector out) {
        // 4. Timer 触发：输出窗口结果
        TimeWindow w = findWindowByEnd(ts);
        Accumulator acc = windowStates.get(w);
        out.collect(emit(acc));
        windowStates.remove(w); // 清理状态
    }
}

stream
    .keyBy(event -> event.userId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new CountAggregate()) // 只需实现 create/add/getResult
    .print();

stream
    .keyBy(event -> event.userId)
    .process(new SessionTimeoutFunction()) // 自定义状态 + Timer
    .print();

// 先用 window + aggregate 做高效聚合
DataStream<ClickStats> stats = clicks
    .keyBy(user)
    .window(...)
    .aggregate(new ClickAgg());

// 再对聚合结果做复杂处理
stats
    .keyBy(stats -> stats.userId)
    .process(new AlertOnSpikesFunction()) // 比如突增告警
    .print();