Apache Flink是一个流处理框架，它提供了许多内置的触发器来控制流处理作业的执行。但是，有时候内置的触发器不能满足我们的需求，这时候我们就需要自定义触发器，在编写自定义触发器之前，我们先来了解一下触发器的基本知识：

一、触发器概述

• 触发器是什么？
       Trigger（触发器）决定了什么时候窗口准备就绪了，一旦窗口准备就绪就可以使用WindowFunction（窗口计算操作）进行计算。每一个 WindowAssigner（窗口分配器） 都会有一个默认的Trigger。如果默认的Trigger不满足用户的需求，用户可以自定义Trigger。
  触发器接口有5个方法，允许触发器对不同的事件做出反应：
  需要注意的是：

1. 前3个方法决定如何响应它们的调用事件，返回一个 TriggerResult：
   CONTINUE：什么也不做
   FIRE：触发计算
   PURGE：清除窗口中的元素
   FIRE_AND_PURGE：触发计算并清空窗口中的元素
2. 以上这些方法都可以用来为之后的操作注册处理时间定时器或事件时间定时器

1. onElement()：每当向窗口添加一个元素时，都会调用该方法。
2. onEventTime()：当注册的事件时间定时器触发时，该方法被调用。
3. onProcessingTime()：当注册的处理时间计时器触发时，该方法被调用。
4. onMerge()：该方法适用于有状态触发器，并在它们对应的窗口合并时合并两个触发器的状态。
5. clear()：执行窗口的清除操作。

1. EventTimeTrigger：基于事件时间和watermark机制来对窗口进行触发计算
2. ProcessingTimeTrigger： 基于处理时间触发
3. CountTrigger：窗口元素数超过预先给定的限制值的话会触发计算
4. PurgingTrigger：作为其它trigger的参数，将其转化为一个purging触发器

二、需求

三、自定义触发器

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.streaming.api.TimeCharacteristic;
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;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public class CountAndSizeTrigger<T> extends Trigger<T, TimeWindow> {

    private static final String DATA_COUNT_STATE_NAME = "dataCountState";

    private static final String DATA_SIZE_STATE_NAME = "dataSizeState";

    // 窗口最大数据条数
    private int maxCount;

    // 窗口最大数据字节数
    private int maxSize;

    // 时间语义：event time、process time
    private TimeCharacteristic timeType;

    // 用于储存窗口当前数据条数的状态对象
    private ReducingStateDescriptor<Long> countStateDescriptor = new ReducingStateDescriptor(DATA_COUNT_STATE_NAME, new ReduceFunction<Long>() {
        @Override
        public Long reduce(Long value1, Long value2) throws Exception {
            return value1 + value2;
        }
    }, LongSerializer.INSTANCE);

    //用于储存窗口当前数据字节数的状态对象
    private ReducingStateDescriptor<Integer> sizeStateDescriptor = new ReducingStateDescriptor(DATA_SIZE_STATE_NAME, new ReduceFunction<Long>() {
        @Override
        public Long reduce(Long value1, Long value2) throws Exception {
            return value1 + value2;
        }
    }, IntSerializer.INSTANCE);

    private CountAndSizeTrigger(int maxCount, int maxSize , TimeCharacteristic timeType) {
        this.maxCount = maxCount;
        this.maxSize = maxSize;
        this.timeType = timeType;
    }

    /**
     * 触发计算，计算结束后清空窗口内的元素
     * @param window 窗口
     * @param ctx 上下文
     */
    private TriggerResult fireAndPurge(TimeWindow window, TriggerContext ctx) throws Exception {
        clear(window, ctx);
        return TriggerResult.FIRE_AND_PURGE;
    }

    /**
     * 进入窗口的每个元素都会调用该方法
     * @param element 元素
     * @param timestamp 时间戳
     * @param window 窗口
     * @param ctx 上下文
     */
    @Override
    public TriggerResult onElement(T element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
        ReducingState<Long> countState = ctx.getPartitionedState(countStateDescriptor);
        ReducingState<Integer> sizeState = ctx.getPartitionedState(sizeStateDescriptor);
        Map<String, JSONArray> map = stringToMap(element.toString());
        if (map != null) {
            for (Map.Entry<String, JSONArray> entry : map.entrySet()) {
                JSONArray value = entry.getValue();
                countState.add(Long.valueOf(value.size()));
            }
        } else {
            countState.add(0L);
        }
        int length = String.valueOf(element).getBytes("utf-8").length;
        sizeState.add(length);
        // 注册定时器
        ctx.registerProcessingTimeTimer(window.maxTimestamp());

        if (countState.get() >= maxCount) {
            log.info("fire count {} ",countState.get());
            return fireAndPurge(window, ctx);
        }if (sizeState.get() >= maxSize){
            log.info("fire size {} ",sizeState.get());
            return fireAndPurge(window, ctx);
        }else{
            return TriggerResult.CONTINUE;
        }
    }

    // 数据处理，可根据需要修改
    private Map<String, JSONArray> stringToMap(String str) {
        if (StringUtils.isBlank(str)) {
            return null;
        }
        String string = str.substring(1, str.length() - 1).replaceAll(" ", "");
        Map<String, JSONArray> map = new HashMap<>();
        String[] split = string.split("=");
        if (split.length < 2) {
            return null;
        } else {
            String key = split[0];
            String value = string.substring(string.indexOf("=") + 1);
            map.put(key, JSON.parseArray(value));
        }
        return map;
    }

    /**
     * 处理时间窗口触发的时候会被调用
     * @param time 时间
     * @param window 窗口
     * @param ctx 上下文
     */
    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
        if (timeType != TimeCharacteristic.ProcessingTime) {
            return TriggerResult.CONTINUE;
        }
        log.info("fire time {} ",time);
        return fireAndPurge(window, ctx);
    }

    /**
     * 事件时间窗口触发的时候被调用
     * @param time 时间
     * @param window 窗口
     * @param ctx 上下文
     */
    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
        if (timeType != TimeCharacteristic.EventTime) {
            return TriggerResult.CONTINUE;
        }
        if (time >= window.getEnd()) {
            return TriggerResult.CONTINUE;
        } else {
            log.info("fire with event tiem: " + time);
            return fireAndPurge(window, ctx);
        }
    }

    /**
     * 执行窗口的清除操作
     * @param window 窗口
     * @param ctx 上下文
     */
    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        ctx.getPartitionedState(countStateDescriptor).clear();
        ctx.getPartitionedState(sizeStateDescriptor).clear();
    }

    /**
     * 初始化触发器，默认使用processTime
     * @param maxCount 最大数据条数
     * @param maxSize 最大数据字节数
     * @param timeType 事件类型
     */
    public static CountAndSizeTrigger creat(int maxCount, int maxSize) {
        return new CountAndSizeTrigger(maxCount,maxSize,TimeCharacteristic.ProcessingTime);
    }

    /**
     * 初始化触发器
     * @param maxCount 最大数据条数
     * @param maxSize 最大数据字节数
     * @param timeType 事件类型
     */
    public static CountAndSizeTrigger creat(int maxCount, int maxSize, TimeCharacteristic timeType) {
        return new CountAndSizeTrigger(maxCount,maxSize,timeType);
    }
}

四、使用示例

stream
        .timeWindowAll(Time.seconds(10))
        .trigger(new CountAndSizeTrigger(1000, 1024))
        .process(new DemoWindowProcessFunction())
        .addSink(new DemoSinkFunction())
        .name("demo");