Flink自定义触发器给出意外输出

Question

val windowedStream = valueStream
                          .keyBy(0)
                          .window(GlobalWindows.create())
                          .trigger(TradeTrigger.of())

@PublicEvolving
public class TradeTrigger<W extends Window> extends Trigger<Object, W> {

    private static final long serialVersionUID = 1L;

    static boolean flag=false;
    static long ctime = System.currentTimeMillis();

    private TradeTrigger() {
    }

    @Override
    public TriggerResult onElement(
            Object arg0,
            long arg1,
            W arg2,
            org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext arg3)
            throws Exception {
        // TODO Auto-generated method stub

        if(flag == false){
            if((System.currentTimeMillis()-ctime) >= 20000){
               flag = true;
               ctime = System.currentTimeMillis();
               return TriggerResult.FIRE;
            }
            return TriggerResult.CONTINUE;
        } else {
            if((System.currentTimeMillis()-ctime) >= 5000){
                ctime = System.currentTimeMillis();
                return TriggerResult.FIRE;
            }
            return TriggerResult.CONTINUE;
        }

    }

    @Override
    public TriggerResult onEventTime(
            long arg0,
            W arg1,
            org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext arg2)
            throws Exception {
        // TODO Auto-generated method stub
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(
            long arg0,
            W arg1,
            org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext arg2)
            throws Exception {
        // TODO Auto-generated method stub
        return TriggerResult.CONTINUE;
    }


    public static <W extends Window> TradeTrigger<W> of() {
        return new TradeTrigger<>();
    }

}

因此，基本上，当标志为假时，即第一次，触发器应在20秒内触发，并将标志设置为真。从下一次开始，它应该每隔5秒就会被发射一次。

我面临的问题是，每次触发器被激发时，输出中只有一条消息。也就是说，我在20秒后收到一条消息，每5秒收到一条消息。我希望在每次触发的输出中有20条消息。

如果我使用.timewindow(time.seconds(5))并创建一个五秒的时间窗口，则每5秒输出20条消息。请帮我把这个代码弄对。我是不是缺了什么？

Answer 1

触发器实现有几个问题：

>

不应将函数的状态存储在静态变量中。Flink不隔离JVM中的用户进程。相反，它使用每个TaskManager一个JVM并启动多个线程。因此，您的静态布尔标志在触发器的多个实例中共享。您应该存储Flink的ValueState接口，该接口可以从TriggerContext访问。闪现会注意检查你的状态，并在失败的情况下恢复它。

trigger.onevent()仅在新事件到达时调用。因此它不能用于在特定时间触发窗口计算。相反，您应该注册事件时间计时器或处理时间计时器（同样通过TriggerContext)。计时器将分别调用Trigger.OnEventTime()或Trigger.OnProcessingTime()。使用事件还是处理时间取决于您的用例。