public class OrderBO {
    /**
     * ID
     */
    private Integer id;
    /***
     * 訂單標(biāo)題
     */
    private String title;
    /**
     * 訂單金額
     */
    private Integer amount;
    /**
     * 訂單狀態(tài)：1-已支付，2-已退款
     */
    private Integer state;
    /**
     * 用戶ID
     */
    private String userId;
}

/**
* 告警處理邏輯
**/
private static class AlarmLogic extends KeyedProcessFunction<String, OrderBO, OrderBO> {
    // 是否已經(jīng)出現(xiàn)退款的標(biāo)記
    private ValueState<Boolean> flagState;
    // 定時(shí)器，時(shí)間到了會(huì)清掉狀態(tài)
    private ValueState<Long> timerState;
    private static final long ONE_MINUTE = 60 * 1000;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Boolean> flagDescriptor = new ValueStateDescriptor<>(
                "flag",
                Types.BOOLEAN);
        flagState = getRuntimeContext().getState(flagDescriptor);

        ValueStateDescriptor<Long> timerDescriptor = new ValueStateDescriptor<>(
                "timer-state",
                Types.LONG);
        timerState = getRuntimeContext().getState(timerDescriptor);
    }

    @Override
    public void processElement(OrderBO value, KeyedProcessFunction<String, OrderBO, OrderBO>.Context context, Collector<OrderBO> collector) throws Exception {
        Boolean refundFlag = flagState.value();

        // 如果已經(jīng)退款過一次了，如果再出現(xiàn)退款則發(fā)射給下個(gè)算子，然后清理掉定時(shí)器。狀態(tài)2代表退款
        if (refundFlag != null && refundFlag) {
            if (value.getState() == 2) {
                collector.collect(value);
            }
            cleanUp(context);
        } else {
            // 如果第一次出現(xiàn)退款，則寫入狀態(tài)，同時(shí)開啟定時(shí)器。狀態(tài)2代表退款
            if (value.getState() == 2) {
                flagState.update(true);
                long timer = context.timerService().currentProcessingTime() + ONE_MINUTE;
                context.timerService().registerProcessingTimeTimer(timer);
                timerState.update(timer);
            }
        }
    }

    /**
     * 定時(shí)器到了之后，清理狀態(tài)值
     */
    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<String, OrderBO, OrderBO>.OnTimerContext ctx, Collector<OrderBO> out) throws Exception {
        timerState.clear();
        flagState.clear();
    }

    /**
     * 手動(dòng)清理狀態(tài)值
     *
     * @param ctx
     * @throws Exception
     */
    private void cleanUp(Context ctx) throws Exception {
        Long timer = timerState.value();
        ctx.timerService().deleteProcessingTimeTimer(timer);

        timerState.clear();
        flagState.clear();
    }
}

public static void main(String[] args) throws Exception {
    // 1、執(zhí)行環(huán)境創(chuàng)建
    StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
    // 2、讀取Socket數(shù)據(jù)端口。實(shí)際根據(jù)具體業(yè)務(wù)對(duì)接數(shù)據(jù)來源
    DataStreamSource<String> orderStream = environment.socketTextStream("localhost", 9527);
    // 3、數(shù)據(jù)讀取個(gè)切割方式
    SingleOutputStreamOperator<OrderBO> resultDataStream = orderStream
            .flatMap(new CleanDataAnd2Order()) // 清洗和處理數(shù)據(jù)
            .keyBy(x -> x.getUserId()) // 分區(qū)
            .process(new AlarmLogic()); // 處理告警邏輯

    // 4、打印分析結(jié)果
    resultDataStream.print("告警===>");
    // 5、環(huán)境啟動(dòng)
    environment.execute("OrderAlarmApp");
}

模擬場(chǎng)景：某個(gè)用戶1分鐘內(nèi)連續(xù)兩次退款，第二次發(fā)出告警。
示例數(shù)據(jù)：
1,aaa,100,1,user1
2,bbb,200,1,user2
3,ccc,300,2,user1
4,ddd,400,2,user1

5,ddd,400,2,user1
6,bbb,200,2,user2
7,bbb,400,2,user2