Flink 自定义Trigger实战代码，结合State来使用

简单说明： 比如window大小为20秒，在进入window的时候 有一个valueState在统计数据数量，数量大于等于2的时候直接触发这个窗口，如果在20秒事件窗口内只有一条数据进入 那么正常等window事件到了 就会触发这一条数据了 。 

package appOperator;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class MyTrigger extends Trigger<JSONObject, TimeWindow> {
    private static final long serialVersionUID = 1L;
    ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("total", Integer.class);
    private static int flag = 0;
    
    private TriggerResult fireAndPurge(TimeWindow window, TriggerContext ctx) throws Exception {
        clear(window, ctx);
        return TriggerResult.FIRE_AND_PURGE;
    }


    @Override
    public TriggerResult onElement(JSONObject element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {

        //todo 注册时间
        ctx.registerProcessingTimeTimer(window.maxTimestamp());
        ValueState<Integer> sumState = ctx.getPartitionedState(stateDescriptor);


       /* if(flag >= 2){
            flag = 0;
            return TriggerResult.FIRE;
        }else{
            flag++;
        }
*/
        if (null == sumState.value()) {
            sumState.update(0);
        }
        int newValue = sumState.value()+1 ;
        sumState.update(newValue);
        System.out.println("newValue = "+newValue+",sumState.value() = "+sumState.value());
        if (sumState.value() >= 2) {
            System.out.println("进入。。。。。。。。。。。。");
            return fireAndPurge(window, ctx);
        }

        return TriggerResult.CONTINUE ;

    }

    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {

        return TriggerResult.FIRE;

    }

    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {

        return TriggerResult.CONTINUE;

 
    }

    @Override
    public void onMerge(TimeWindow window,
                        OnMergeContext ctx) {

        long windowMaxTimestamp = window.maxTimestamp();
        if (windowMaxTimestamp > ctx.getCurrentProcessingTime()) {
            ctx.registerProcessingTimeTimer(windowMaxTimestamp);
        }
    }

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        ValueState<Integer> partitionedState = ctx.getPartitionedState(stateDescriptor);
        partitionedState.clear();
        ctx.deleteProcessingTimeTimer(window.maxTimestamp());
    }
} 

调用的代码：