java flink（二十五）实战之电商黑名单过滤使用：控制定时器触发、删除。实现：1、根据时间触发定时器 2、根据事件触发定时器

谁借莪１个温暖的怀抱￠ 2023-01-22 15:49 1阅读 0赞

**题目：**

1、我们获取访问日志，如果2s内连续登录失败超过两次的用户进行报警

2、解决时效性 连续失败等于两次就报警

**1、目录结构：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70][]

**2、日志内容：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1][]

**3、接收数据包装类**

package Beans;
    
    public class LoginEvent {
        private Long userId;
        private String ip;
        private String loginState;
        private Long timestamp;
    
        public LoginEvent() {
        }
    
        public LoginEvent(Long userId, String ip, String loginState, Long timestamp) {
            this.userId = userId;
            this.ip = ip;
            this.loginState = loginState;
            this.timestamp = timestamp;
        }
    
        public Long getUserId() {
            return userId;
        }
    
        public void setUserId(Long userId) {
            this.userId = userId;
        }
    
        public String getIp() {
            return ip;
        }
    
        public void setIp(String ip) {
            this.ip = ip;
        }
    
        public String getLoginState() {
            return loginState;
        }
    
        public void setLoginState(String loginState) {
            this.loginState = loginState;
        }
    
        public Long getTimestamp() {
            return timestamp;
        }
    
        public void setTimestamp(Long timestamp) {
            this.timestamp = timestamp;
        }
    
        @Override
        public String toString() {
            return "LoginEvent{" +
                    "userId=" + userId +
                    ", ip='" + ip + '\'' +
                    ", loginState='" + loginState + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }

输出报警包装类：

package Beans;
    
    public class LoginFailWarning {
        private Long userId;
        private Long firstFailTime;
        private Long lastFailTime;
        private String warningMsg;
    
        public LoginFailWarning() {
        }
    
        public LoginFailWarning(Long userId, Long firstFailTime, Long lastFailTime, String warningMsg) {
            this.userId = userId;
            this.firstFailTime = firstFailTime;
            this.lastFailTime = lastFailTime;
            this.warningMsg = warningMsg;
        }
    
        public Long getUserId() {
            return userId;
        }
    
        public void setUserId(Long userId) {
            this.userId = userId;
        }
    
        public Long getFirstFailTime() {
            return firstFailTime;
        }
    
        public void setFirstFailTime(Long firstFailTime) {
            this.firstFailTime = firstFailTime;
        }
    
        public Long getLastFailTime() {
            return lastFailTime;
        }
    
        public void setLastFailTime(Long lastFailTime) {
            this.lastFailTime = lastFailTime;
        }
    
        public String getWarningMsg() {
            return warningMsg;
        }
    
        public void setWarningMsg(String warningMsg) {
            this.warningMsg = warningMsg;
        }
    
        @Override
        public String toString() {
            return "LoginFailWarning{" +
                    "userId=" + userId +
                    ", firstFailTime=" + firstFailTime +
                    ", lastFailTime=" + lastFailTime +
                    ", warningMsg='" + warningMsg + '\'' +
                    '}';
        }
    }

**4、 2s内连续失败超过两次 触发定时器报警，如果成功删除定时器**

package Project;
    
    import Beans.LoginEvent;
    import Beans.LoginFailWarning;
    import org.apache.flink.api.common.state.ListState;
    import org.apache.flink.api.common.state.ListStateDescriptor;
    import org.apache.flink.api.common.state.ValueState;
    import org.apache.flink.api.common.state.ValueStateDescriptor;
    import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.TimeCharacteristic;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.util.Collector;
    
    import java.net.URL;
    import java.sql.DatabaseMetaData;
    import java.util.ArrayList;
    
    public class LoginFail {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    
            //读取数据
            URL resource = LoginFail.class.getResource("/LoginLog.csv");
            DataStream<LoginEvent> logEventStream = env.readTextFile(resource.getPath())
                    .map(line -> {
                        String[] fields = line.split(",");
                        return new LoginEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
                    })
                    .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LoginEvent>(Time.seconds(3)) {
                        @Override
                        public long extractTimestamp(LoginEvent loginEvent) {
                            return loginEvent.getTimestamp() * 1000;
                        }
                    });
            //自定义处理函数
            SingleOutputStreamOperator<LoginFailWarning> warningStream = logEventStream.keyBy(LoginEvent::getUserId)
                    .process(new LoginFailDetectWarning(2));
            warningStream.print();
            env.execute("login fail detect job");
        }
        //实现自定义keydprocessfunction
        public static class LoginFailDetectWarning extends KeyedProcessFunction<Long, LoginEvent, LoginFailWarning>{
            //连续登录失败次数
            private Integer maxFailTimes;
    
            public LoginFailDetectWarning(Integer maxFailTimes) {
                this.maxFailTimes = maxFailTimes;
            }
    
            //定义状态 保存2s内所有失败记录
            ListState<LoginEvent> loginFailEventListState;
            //定义状态 保存注册的定时器的时间戳
            ValueState<Long> timerTsState;
    
            @Override
            public void open(Configuration parameters) throws Exception {
                loginFailEventListState = getRuntimeContext().getListState(new ListStateDescriptor<LoginEvent>("login-fail-list",LoginEvent.class));
                timerTsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-ts", Long.class));
            }
    
            @Override
            public void processElement(LoginEvent loginEvent, Context context, Collector<LoginFailWarning> collector) throws Exception {
                //判断当前登录事件类型
                if ("fail".equals(loginEvent.getLoginState())){
                    //如果是失败时间 添加到列表状态
                    loginFailEventListState.add(loginEvent);
                    //如果没有定时器 注册一个2s之后的定时器
                    if(timerTsState.value()==null){
                        Long ts = (loginEvent.getTimestamp()+2)*1000;
                        context.timerService().registerEventTimeTimer(ts);
                        timerTsState.update(ts);
                    }
                }else{
                    //如果是登陆成功 删除定时器 清空状态 重新开始
                    if (timerTsState.value()!=null)
                        context.timerService().deleteProcessingTimeTimer(timerTsState.value());
                        loginFailEventListState.clear();
                        timerTsState.clear();
                    }
            }
    
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<LoginFailWarning> out) throws Exception {
                //定时器触发  说明2s内没有成功事件进来 判断list中失败的个数
                ArrayList<LoginEvent> loginFailEvents = Lists.newArrayList(loginFailEventListState.get());
                Integer failTimes = loginFailEvents.size();
                if(failTimes>=maxFailTimes){
                    //如果超过设定的次数 输出报警
                    out.collect(new LoginFailWarning(ctx.getCurrentKey(),loginFailEvents.get(0).getTimestamp(),loginFailEvents.get(failTimes-1).getTimestamp(),"login fail in 2s for "+ failTimes+" times."));
                }
                //清空状态
                loginFailEventListState.clear();
                timerTsState.clear();
            }
        }
    }

事件报警：

package Project;
    
    import Beans.LoginEvent;
    import Beans.LoginFailWarning;
    import org.apache.flink.api.common.state.ListState;
    import org.apache.flink.api.common.state.ListStateDescriptor;
    import org.apache.flink.api.common.state.ValueState;
    import org.apache.flink.api.common.state.ValueStateDescriptor;
    import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.TimeCharacteristic;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.util.Collector;
    
    import java.net.URL;
    import java.sql.DatabaseMetaData;
    import java.util.ArrayList;
    import java.util.Iterator;
    
    public class LoginFail {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    
            //读取数据
            URL resource = LoginFail.class.getResource("/LoginLog.csv");
            DataStream<LoginEvent> logEventStream = env.readTextFile(resource.getPath())
                    .map(line -> {
                        String[] fields = line.split(",");
                        return new LoginEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
                    })
                    .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LoginEvent>(Time.seconds(3)) {
                        @Override
                        public long extractTimestamp(LoginEvent loginEvent) {
                            return loginEvent.getTimestamp() * 1000;
                        }
                    });
            //自定义处理函数
            SingleOutputStreamOperator<LoginFailWarning> warningStream = logEventStream.keyBy(LoginEvent::getUserId)
                    .process(new LoginFailDetectWarning(2));
            warningStream.print();
            env.execute("login fail detect job");
        }
        //实现自定义keydprocessfunction
        public static class LoginFailDetectWarning extends KeyedProcessFunction<Long, LoginEvent, LoginFailWarning>{
            //连续登录失败次数
            private Integer maxFailTimes;
    
            public LoginFailDetectWarning(Integer maxFailTimes) {
                this.maxFailTimes = maxFailTimes;
            }
    
            //定义状态 保存2s内所有失败记录
            ListState<LoginEvent> loginFailEventListState;
    
            @Override
            public void open(Configuration parameters) throws Exception {
                loginFailEventListState = getRuntimeContext().getListState(new ListStateDescriptor<LoginEvent>("login-fail-list",LoginEvent.class));
    
            }
    
            //以登录事件作为判断报警的出发条件 不再注册定时器
            @Override
            public void processElement(LoginEvent loginEvent, Context context, Collector<LoginFailWarning> collector) throws Exception {
               //判断当前事件登录状态
                if("fail".equals(loginEvent.getLoginState())){
                    //如果是登录失败 继续获取状态中之前的登录失败事件 继续判断是否已有失败登录事件
                    Iterator<LoginEvent> iterator = loginFailEventListState.get().iterator();
                    if(iterator.hasNext()){
                        //如果已经有登录失败事件 判断是否在2s之内
                        //获取已有的登录失败事件
                        LoginEvent firstFailEvent = iterator.next();
                        if(loginEvent.getTimestamp()-firstFailEvent.getTimestamp()<=2){
                            collector.collect(new LoginFailWarning(loginEvent.getUserId(),firstFailEvent.getTimestamp(),loginEvent.getTimestamp(),"login fail in 2s"));
                        }
                        //不管报不报警  都处理完毕 更新状态
                        loginFailEventListState.clear();
                        loginFailEventListState.add(loginEvent);
                    }else{
                        //如果没有登录失败 直接将当前事件存入
                        loginFailEventListState.add(loginEvent);
                    }
    
                }else{
                    //如果是登录成功 直接清空状态
                    loginFailEventListState.clear();
                }
            }
    
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<LoginFailWarning> out) throws Exception {
    
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70]: /images/20221020/c8e9fa75ae4045cd9aaa557c4922d6c1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1]: /images/20221020/e6fbaa19ebfb4d918707f061634f4b7f.png