开发前端软件素材源码教程下载网站

全部
网站模板
网页特效
网站源码
素材资源
建站学院
软件工具
mysql mysql 微信 微信 小程序 苹果 dedecms 修真图
当前位置：首页 > 建站学院 > 其他教程 >  基于FLink如何实现实时安全检测

基于FLink如何实现实时安全检测

其他教程   发布日期：2023年07月04日   浏览次数：501次

这篇“基于FLink如何实现实时安全检测”文章的知识点大部分人都不太理解，所以小编给大家总结了以下内容，内容详细，步骤清晰，具有一定的借鉴价值，希望大家阅读完这篇文章能有所收获，下面我们一起来看看这篇“基于FLink如何实现实时安全检测”文章吧。

场景描述

针对一个内部系统，如邮件系统，公司员工的访问行为日志，存放于kafka，我们希望对于一个用户账号在同一个IP下，任意的3分钟时间内，连续登录邮件系统20次失败，下一次登录成功，这种场景能够及时获取并推送到企业微信某个指定的安全接口人。kafka中的数据，能够通过某个关键字，区分当前网络访问是否一次登录事件，且有访问时间（也就是事件时间）。在解析到符合需求的用户账号之后，第一时间进行企业微信告警推送，并将其这段时间内的访问行为，写入下游ElasticSearch。

组件版本

  • Flink-1.14.4

  • Java8

  • ElasticSearch-7.3.2

  • Kafka-2.12_2.8.1

日志结构

IP和账号皆为测试使用。

  1. {
  2.    "user": "wangxm",
  3.    "client_ip": "110.68.6.182",
  4.    "source": "login",
  5.    "loginname": "wangxm@test.com",
  6.    "IP": "110.8.148.58",
  7.    "timestamp": "17:58:12",
  8.    "@timestamp": "2022-04-20T09:58:13.647Z",
  9.    "ip": "110.7.231.25",
  10.    "clienttype": "POP3",
  11.    "result": "success",
  12.    "@version": "1"
  13.  }

技术方案

上述场景，可考虑使用FlinkCEP及Flink的滑动窗口进行实现。由于本人在采用FlinkCEP的方案进行代码编写调试后，发现并不能满足，因此改用滑动窗口进行实现。

关键代码

主入口类

主入口类，创建了flink环境、设置了基础参数，创建了kafkaSource,接入消息后，进行了映射、过滤，并设置了水位线，进行了分组，之后设置了滑动窗口，在窗口内进行了事件统计，将复合条件的事件收集返回并写入ElasticSearch。

针对map、filter、keyBy、window等算子，都单独进行了编写，后面会一一列出来。

  1. package com.data.dev.flink.mailTopic.main;
  3. import com.data.dev.common.javabean.BaseBean;
  4. import com.data.dev.common.javabean.kafkaMailTopic.MailMsgAlarm;
  5. import com.data.dev.elasticsearch.ElasticSearchInfo;
  6. import com.data.dev.elasticsearch.SinkToEs;
  7. import com.data.dev.flink.FlinkEnv;
  8. import com.data.dev.flink.mailTopic.OperationForLoginFailCheck.*;
  9. import com.data.dev.kafka.KafkaSourceBuilder;
  10. import com.data.dev.key.ConfigurationKey;
  11. import com.data.dev.utils.TimeUtils;
  12. import lombok.extern.slf4j.Slf4j;
  13. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  14. import org.apache.flink.connector.kafka.source.KafkaSource;
  15. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  16. import org.apache.flink.streaming.api.datastream.KeyedStream;
  17. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  18. import org.apache.flink.streaming.api.datastream.WindowedStream;
  19. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  20. import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
  21. import org.apache.flink.streaming.api.windowing.time.Time;
  22. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  24. import java.time.Duration;
  27. /**
  28.  * Flink处理在3分钟内连续登录失败20次后登录成功的场景
  29.  * 采用滑动窗口来实现
  30.  * @author wangxiaomin 2022-06-01
  31.  */
  33. @Slf4j
  34. public class MailMsg extends BaseBean {
  36.     /**
  37.      * Flink作业名称
  38.      */
  39.     public static final  String JobName = "告警采集平台――连续登录失败后登录成功告警";
  40.     /**
  41.      * Kafka消息名
  42.      */
  43.     public static final  String KafkaSourceName = "Kafka Source for AlarmPlatform About Mail Topic";
  45.     public MailMsg(){
  46.         log.info("初始化滑动窗口场景告警程序");
  47.     }
  49.     /**
  50.      * 执行逻辑统计场景，实现告警推送
  51.      */
  52.     public static void execute(){
  55.         //① 创建Flink执行环境并设置checkpoint等必要的参数
  56.         StreamExecutionEnvironment env = FlinkEnv.getFlinkEnv();
  57.         KafkaSource<String> kafkaSource = KafkaSourceBuilder.getKafkaSource(ConfigurationKey.KAFKA_MAIL_TOPIC_NAME,ConfigurationKey.KAFKA_MAIL_CONSUMER_GROUP_ID) ;
  58.         DataStreamSource<String> kafkaMailMsg = env.fromSource(kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofMillis(10)), KafkaSourceName);
  61.         //② 筛选登录消息，创建初始登录事件流
  62.         SingleOutputStreamOperator<com.data.dev.common.javabean.kafkaMailTopic.MailMsg> loginMapDs = kafkaMailMsg.map(new MsgToBeanMapper()).name("Map算子加工");
  63.         SingleOutputStreamOperator<com.data.dev.common.javabean.kafkaMailTopic.MailMsg> loginFilterDs = loginMapDs.filter(new MailMsgForLoginFilter()).name("Filter算子加工");
  65.         //③ 设置水位线
  66.         WatermarkStrategy<com.data.dev.common.javabean.kafkaMailTopic.MailMsg> watermarkStrategy = WatermarkStrategy.<com.data.dev.common.javabean.kafkaMailTopic.MailMsg>forBoundedOutOfOrderness(Duration.ofMinutes(1))
  67.                         .withTimestampAssigner((mailMsg, timestamp) -> TimeUtils.switchUTCToBeijingTimestamp(mailMsg.getTimestamp_datetime()));
  68.         SingleOutputStreamOperator<com.data.dev.common.javabean.kafkaMailTopic.MailMsg> loginWmDs = loginFilterDs.assignTimestampsAndWatermarks(watermarkStrategy.withIdleness(Duration.ofMinutes(3))).name("增加水位线");
  70.         //④ 设置主键
  71.         KeyedStream<com.data.dev.common.javabean.kafkaMailTopic.MailMsg, String> loginKeyedDs = loginWmDs.keyBy(new LoginKeySelector());
  73.         //⑥ 转化为滑动窗口
  74.         WindowedStream<com.data.dev.common.javabean.kafkaMailTopic.MailMsg, String, TimeWindow> loginWindowDs = loginKeyedDs.window(SlidingEventTimeWindows.of(Time.seconds(180L),Time.seconds(90L)));
  76.         //⑦ 在窗口内进行逻辑统计
  77.         SingleOutputStreamOperator<MailMsgAlarm> loginWindowsDealDs  = loginWindowDs.process(new WindowProcessFuncImpl()).name("窗口处理逻辑");
  79.         //⑧ 将结果转化为通用DataStream<String>格式
  80.         SingleOutputStreamOperator<String> resultDs  = loginWindowsDealDs.map(new AlarmMsgToStringMapper()).name("窗口结果转化为标准格式");
  82.         //⑨ 将最终结果写入ES
  83.         resultDs.addSink(SinkToEs.getEsSinkBuilder(ElasticSearchInfo.ES_LOGIN_FAIL_INDEX_NAME,ElasticSearchInfo.ES_INDEX_TYPE_DEFAULT).build());
  85.         //⑩ 提交Flink集群进行执行
  86.         FlinkEnv.envExec(env,JobName);
  88.     }
  89. }

mapper算子

  1. package com.data.dev.flink.mailTopic.OperationForLoginFailCheck;
  3. import com.alibaba.fastjson.JSON;
  4. import com.data.dev.common.javabean.BaseBean;
  5. import com.data.dev.common.javabean.kafkaMailTopic.MailMsgAlarm;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.apache.flink.api.common.functions.MapFunction;
  9. /**
  10.  *  逻辑统计场景告警推送ES消息体
  11.  *  @author wangxiaoming-ghq 2022-06-01
  12.  */
  13. @Slf4j
  14. public   class AlarmMsgToStringMapper extends BaseBean implements MapFunction<MailMsgAlarm, String> {
  16.     @Override
  17.     public String map(MailMsgAlarm mailMsgAlarm) throws Exception {
  18.         return JSON.toJSONString(mailMsgAlarm);
  19.     }
  20. }

filter算子

  1. package com.data.dev.flink.mailTopic.OperationForLoginFailCheck;
  3. import com.data.dev.common.javabean.BaseBean;
  4. import com.data.dev.common.javabean.kafkaMailTopic.MailMsg;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.apache.flink.api.common.functions.FilterFunction;
  9. /**
  10.  * ②　消费mail主题的消息，过滤其中login的事件
  11.  * @author wangxiaoming-ghq 2022-06-01
  12.  */
  13. @Slf4j
  14. public class MailMsgForLoginFilter extends BaseBean implements FilterFunction<MailMsg> {
  15.     @Override
  16.     public boolean filter(MailMsg mailMsg) {
  17.         if("login".equals(mailMsg.getSource())) {
  18.             log.info("筛选原始的login事件：【" + mailMsg + "】");
  19.         }
  20.         return "login".equals(mailMsg.getSource());
  21.     }
  22. }

keyBy算子

  1. package com.data.dev.flink.mailTopic.OperationForLoginFailCheck;
  3. import com.data.dev.common.javabean.BaseBean;
  4. import com.data.dev.common.javabean.kafkaMailTopic.MailMsg;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.apache.flink.api.java.functions.KeySelector;
  8. /**
  9.  * CEP 编程，需要进行key选取
  10.  */
  11. @Slf4j
  12. public class LoginKeySelector extends BaseBean implements KeySelector<MailMsg, String> {
  13.     @Override
  14.     public String getKey(MailMsg mailMsg) {
  15.         return mailMsg.getUser() + "@" + mailMsg.getClient_ip();
  16.     }
  17. }

窗口函数（核心代码）

这里我们主要考虑使用一个事件列表，用来存储每一个窗口期内得到的连续登录，当检测到登陆失败的事件，即存入事件列表中，之后判断下一次登录失败事件，如果检测到登录成功事件，但此时登录失败的次数不足20次，则清空loginEventList，等待下一次检测。一旦符合窗口内连续登录失败超过20次且下一次登录成功这个事件，则清空此时的loginEventList并将当前登录成功的事件进行告警推送。

  1. package com.data.dev.flink.mailTopic.OperationForLoginFailCheck;
  3. import com.data.dev.common.javabean.kafkaMailTopic.MailMsg;
  4. import com.data.dev.common.javabean.kafkaMailTopic.MailMsgAlarm;
  5. import com.data.dev.utils.HttpUtils;
  6. import com.data.dev.utils.IPUtils;
  7. import lombok.extern.slf4j.Slf4j;
  8. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  9. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  10. import org.apache.flink.util.Collector;
  12. import java.io.Serializable;
  13. import java.util.ArrayList;
  14. import java.util.List;
  16. /**
  17.  *  滑动窗口内复杂事件解析逻辑实现
  18.  *  @author wangxiaoming-ghq 2022-06-01
  19.  */
  20. @Slf4j
  21. public   class WindowProcessFuncImpl extends  ProcessWindowFunction<MailMsg, MailMsgAlarm, String, TimeWindow> implements Serializable {
  22.     @Override
  23.     public void process(String key, ProcessWindowFunction<MailMsg, MailMsgAlarm, String, TimeWindow>.Context context, Iterable<MailMsg> iterable, Collector<MailMsgAlarm> collector) {
  25.         List<MailMsg> loginEventList = new ArrayList<>();
  26.         MailMsgAlarm mailMsgAlarm;
  27.         for (MailMsg mailMsg : iterable) {
  28.             log.info("收集到的登录事件【" + mailMsg + "】");
  30.             if (mailMsg.getResult().equals("fail")) { //开始检测当前窗口内的事件，并将失败的事件收集到loginEventList
  31.                 log.info("开始检测当前窗口内的事件，并将失败的事件收集到loginEventList");
  32.                 loginEventList.add(mailMsg);
  33.             } else if (mailMsg.getResult().equals("success") && loginEventList.size() < 20) {//如果检测到登录成功事件，但此时登录失败的次数不足20次，则清空loginEventList，等待下一次检测
  34.                 log.info("检测到登录成功事件，但此时登录失败的次数为【" + loginEventList.size() + "】不足20次，清空loginEventList，等待下一次检测");
  35.                 loginEventList.clear();
  36.             } else if (mailMsg.getResult().equals("success") && loginEventList.size() >= 20) {
  37.                 mailMsgAlarm = getMailMsgAlarm(loginEventList,mailMsg);
  38.                 log.info("检测到登录成功的事件,此时窗口内连续登录失败的次数为【" + mailMsgAlarm.getFailTimes() + "】");
  40.                 //一旦符合窗口内连续登录失败超过20次且下一次登录成功这个事件，则清空此时的loginEventList并将当前登录成功的事件进行告警推送；
  41.                 loginEventList.clear();
  42.                 doAlarmPush(mailMsgAlarm);
  44.                 collector.collect(mailMsgAlarm);//将当前登录成功的事件进行收集上报
  45.             } else {
  46.                 log.info(mailMsg.getUser() + "当前已连续：【" + loginEventList.size() + "】 次登录失败");
  47.             }
  48.         }
  49.     }
  52.     /**
  53.      * 2022年6月17日15:03:06
  54.      * @param eventList:当前窗口内的事件列表
  55.      * @param eventCurrent：当前登录成功的事件
  56.      * @return mailMsgAlarm：告警消息体
  57.      */
  58.     public static MailMsgAlarm getMailMsgAlarm(List<MailMsg> eventList,MailMsg eventCurrent){
  60.         String alarmKey = eventCurrent.getUser() + "@" + eventCurrent.getClient_ip();
  61.         String loginFailStartTime = eventList.get(0).getTimestamp_datetime();
  62.         String loginSuccessTime = eventCurrent.getTimestamp_datetime();
  63.         int loginFailTimes = eventList.size();
  65.         MailMsgAlarm mailMsgAlarm = new MailMsgAlarm();
  66.         mailMsgAlarm.setMailMsg(eventCurrent);
  67.         mailMsgAlarm.setAlarmKey(alarmKey);
  68.         mailMsgAlarm.setStartTime(loginFailStartTime);
  69.         mailMsgAlarm.setEndTime(loginSuccessTime);
  70.         mailMsgAlarm.setFailTimes(loginFailTimes);
  72.         return mailMsgAlarm;
  73.     }
  75.     /**
  76.      * 2022年6月17日14:47:53
  77.      * @param mailMsgAlarm :当前构建的需要告警的事件
  78.      */
  79.     public void doAlarmPush(MailMsgAlarm mailMsgAlarm){
  80.         String userKey = mailMsgAlarm.getAlarmKey();
  81.         String clientIp = mailMsgAlarm.mailMsg.getClient_ip();
  82.         boolean isWhiteListIp = IPUtils.isWhiteListIp(clientIp);
  83.         if(isWhiteListIp){//如果是白名单IP，不告警
  84.             log.info("当前登录用户【" + userKey + "】属于白名单IP");
  85.         }else {
  86.             //IP归属查询结果、企业微信推送告警
  87.             String user = HttpUtils.getUserByClientIp(clientIp);
  88.             HttpUtils.pushAlarmMsgToWechatWork(user,mailMsgAlarm.toString());
  89.         }
  90.     }
  91. }

最后一次map算子

  1. package com.data.dev.flink.mailTopic.OperationForLoginFailCheck;
  3. import com.alibaba.fastjson.JSON;
  4. import com.data.dev.common.javabean.BaseBean;
  5. import com.data.dev.common.javabean.kafkaMailTopic.MailMsgAlarm;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.apache.flink.api.common.functions.MapFunction;
  9. /**
  10.  *  逻辑统计场景告警推送ES消息体
  11.  *  @author wangxiaoming-ghq 2022-06-01
  12.  */
  13. @Slf4j
  14. public   class AlarmMsgToStringMapper extends BaseBean implements MapFunction<MailMsgAlarm, String> {
  16.     @Override
  17.     public String map(MailMsgAlarm mailMsgAlarm) throws Exception {
  18.         return JSON.toJSONString(mailMsgAlarm);
  19.     }
  20. }

ElasticSearch工具类

  1. package com.data.dev.elasticsearch;
  3. import com.data.dev.common.javabean.BaseBean;
  4. import com.data.dev.key.ConfigurationKey;
  5. import com.data.dev.key.ElasticSearchKey;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.apache.flink.api.common.functions.RuntimeContext;
  8. import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
  9. import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
  10. import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
  11. import org.apache.flink.streaming.connectors.elasticsearch7.RestClientFactory;
  12. import org.apache.http.HttpHost;
  13. import org.apache.http.auth.AuthScope;
  14. import org.apache.http.auth.UsernamePasswordCredentials;
  15. import org.apache.http.client.CredentialsProvider;
  16. import org.apache.http.impl.client.BasicCredentialsProvider;
  17. import org.elasticsearch.action.index.IndexRequest;
  18. import org.elasticsearch.client.Requests;
  20. import java.util.ArrayList;
  21. import java.util.HashMap;
  22. import java.util.List;
  23. import java.util.Map;
  25. /**
  26.  * 2022年6月17日15:15:06
  27.  * @author wangxiaoming-ghq
  28.  * Flink流计算结果写入ES公共方法
  29.  */
  30. @Slf4j
  31. public class SinkToEs extends BaseBean {
  32.     public static final long serialVersionUID = 2L;
  33.     private static final HashMap<String,String> ES_PROPS_MAP = ConfigurationKey.getApplicationProps();
  34.     private static final String HOST = ES_PROPS_MAP.get(ConfigurationKey.ES_HOST);
  35.     private static final String PASSWORD = ES_PROPS_MAP.get(ConfigurationKey.ES_PASSWORD);
  36.     private static final String USERNAME = ES_PROPS_MAP.get(ConfigurationKey.ES_USERNAME);
  37.     private static final String PORT = ES_PROPS_MAP.get(ConfigurationKey.ES_PORT);
  39.     /**
  40.      * 2022年6月17日15:17:55
  41.      * 获取ES连接信息
  42.      * @return esInfoMap:ES连接信息持久化
  43.      */
  44.     public static HashMap<String,String > getElasticSearchInfo(){
  45.         log.info("获取ES连接信息：【 " + "HOST="+HOST + "PORT="+PORT+"USERNAME="+USERNAME+"PASSWORD=********" + " 】");
  46.         HashMap<String,String> esInfoMap = new HashMap<>();
  47.         esInfoMap.put(ElasticSearchKey.HOST,HOST);
  48.         esInfoMap.put(ElasticSearchKey.PASSWORD,PASSWORD);
  49.         esInfoMap.put(ElasticSearchKey.USERNAME,USERNAME);
  50.         esInfoMap.put(ElasticSearchKey.PORT,PORT);
  52.         return esInfoMap;
  53.     }
  55.     /**
  56.      * @param esIndexName:写入索引名称
  57.      * @param esType：写入索引类型
  58.      * @return ElasticsearchSink.Builder<String>：构建器
  59.      */
  60.     public static ElasticsearchSink.Builder<String> getEsSinkBuilder(String esIndexName,String esType){
  61.         HashMap<String, String> esInfoMap = getElasticSearchInfo();
  62.         List<HttpHost> httpHosts = new ArrayList<>();
  63.         httpHosts.add(new HttpHost(String.valueOf(esInfoMap.get(ElasticSearchKey.HOST)), Integer.parseInt(esInfoMap.get(ElasticSearchKey.PORT)), "http"));
  65.         ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
  66.                 httpHosts,
  67.                 new ElasticsearchSinkFunction<String>() {
  69.                     public IndexRequest createIndexRequest() {
  70.                         Map<String, String> json = new HashMap<>();
  71.                         //log.info("写入ES的data:【"+json+"】");
  72.                         IndexRequest index  = Requests.indexRequest()
  73.                                 .index(esIndexName)
  74.                                 .type(esType)
  75.                                 .source(json);
  76.                         return index;
  77.                     }
  79.                     @Override
  80.                     public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
  81.                         indexer.add(createIndexRequest());
  82.                     }
  83.                 }
  84.         );
  87.         //定义es的连接配置  带用户名密码
  88.         RestClientFactory restClientFactory = restClientBuilder -> {
  89.             CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
  90.             credentialsProvider.setCredentials(
  91.                     AuthScope.ANY,
  92.                     new UsernamePasswordCredentials(
  93.                             String.valueOf(esInfoMap.get(ElasticSearchKey.USERNAME)),
  94.                             String.valueOf(esInfoMap.get(ElasticSearchKey.PASSWORD))
  95.                     )
  96.             );
  97.             restClientBuilder.setHttpClientConfigCallback(httpAsyncClientBuilder -> {
  98.                 httpAsyncClientBuilder.disableAuthCaching();
  99.                 return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
  100.             });
  101.         };
  103.         esSinkBuilder.setRestClientFactory(restClientFactory);
  104.         return esSinkBuilder;
  105.     }
  107. }

事件实体类

  1. package com.data.dev.common.javabean.kafkaMailTopic;
  3. import com.data.dev.common.javabean.BaseBean;
  4. import lombok.Data;
  6. import java.util.Objects;
  9. /**
  10.  * @author wangxiaoming-ghq 2022-05-15
  11.  * 逻辑统计场景告警事件
  12.  */
  13. @Data
  14. public class MailMsgAlarm extends BaseBean {
  17.     /**
  18.      * 当前登录成功的事件
  19.      */
  20.    public  MailMsg mailMsg;
  22.     /**
  23.      * 当前捕获的告警主键：username@client_ip
  24.      */
  25.    public  String alarmKey;
  27.     /**
  28.      * 第一次登录失败的事件时间
  29.      */
  30.    public  String startTime;
  32.     /**
  33.      * 连续登录失败后下一次登录成功的事件时间
  34.      */
  35.    public  String endTime;
  37.     /**
  38.      * 连续登录失败的次数
  39.      */
  40.    public  int failTimes;
  42.     @Override
  43.     public String toString() {
  44.         return "{" +
  45.                 "  'mailMsg_login_success':'" + mailMsg + "'" +
  46.                 ", 'alarmKey':'" + alarmKey + "'" +
  47.                 ", 'start_login_time_in3min':'"  +startTime + "'" +
  48.                 ", 'end_login_time_in3min':'"  +endTime + "'" +
  49.                 ", 'login_fail_times':'"  +failTimes +  "'" +
  50.                 "}";
  51.     }
  53.     public MailMsgAlarm() {
  54.     }
  56.     @Override
  57.     public boolean equals(Object o) {
  58.         if (this == o) return true;
  59.         if (!(o instanceof MailMsgAlarm)) return false;
  60.         MailMsgAlarm that = (MailMsgAlarm) o;
  61.         return getFailTimes() == that.getFailTimes() && getMailMsg().equals(that.getMailMsg()) && getAlarmKey().equals(that.getAlarmKey()) && getStartTime().equals(that.getStartTime()) && getEndTime().equals(that.getEndTime());
  62.     }
  64.     @Override
  65.     public int hashCode() {
  66.         return Objects.hash(getMailMsg(), getAlarmKey(), getStartTime(), getEndTime(), getFailTimes());
  67.     }
  68. }

消息实体类

  1. package com.data.dev.common.javabean.kafkaMailTopic;
  3. import com.data.dev.common.javabean.BaseBean;
  4. import lombok.Data;
  6. import java.util.Objects;
  8. /**
  9.  * {
  10.  *   "user": "wangxm",
  11.  *   "client_ip": "110.68.6.182",
  12.  *   "source": "login",
  13.  *   "loginname": "wangxm@test.com",
  14.  *   "IP": "110.8.148.58",
  15.  *   "timestamp": "17:58:12",
  16.  *   "@timestamp": "2022-04-20T09:58:13.647Z",
  17.  *   "ip": "110.7.231.25",
  18.  *   "clienttype": "POP3",
  19.  *   "result": "success",
  20.  *   "@version": "1"
  21.  * }
  22.  *
  23.  * user登录用户
  24.  * client_ip 来源ip
  25.  * source 类型
  26.  * loginname 登录用户邮箱地址
  27.  * ip 目标前端ip
  28.  * timestamp 发送时间
  29.  * &#064;timestamp  发送日期时间
  30.  * IP 邮件日志发送来源IP
  31.  * clienttype 客户端登录类型
  32.  * result 登录状态
  33.  */
  35. @Data
  36. public class MailMsg extends BaseBean {
  37.     public String user;
  38.     public String client_ip;
  39.     public String source;
  40.     public String loginName;
  41.     public String mailSenderSourceIp;
  42.     public String timestamp_time;
  43.     public String timestamp_datetime;
  44.     public String ip;
  45.     public String clientType;
  46.     public String result;
  47.     public String version;
  49.     public MailMsg() {
  50.     }
  52.     public MailMsg(String user, String client_ip, String source, String loginName, String mailSenderSourceIp, String timestamp_time, String timestamp_datetime, String ip, String clientType, String result, String version) {
  53.         this.user = user;
  54.         this.client_ip = client_ip;
  55.         this.source = source;
  56.         this.loginName = loginName;
  57.         this.mailSenderSourceIp = mailSenderSourceIp;
  58.         this.timestamp_time = timestamp_time;
  59.         this.timestamp_datetime = timestamp_datetime;
  60.         this.ip = ip;
  61.         this.clientType = clientType;
  62.         this.result = result;
  63.         this.version = version;
  64.     }
  66.     @Override
  67.     public boolean equals(Object o) {
  68.         if (this == o) return true;
  69.         if (!(o instanceof MailMsg)) return false;
  70.         MailMsg mailMsg = (MailMsg) o;
  71.         return getUser().equals(mailMsg.getUser()) && getClient_ip().equals(mailMsg.getClient_ip()) && getSource().equals(mailMsg.getSource()) && getLoginName().equals(mailMsg.getLoginName()) && getMailSenderSourceIp().equals(mailMsg.getMailSenderSourceIp()) && getTimestamp_time().equals(mailMsg.getTimestamp_time()) && getTimestamp_datetime().equals(mailMsg.getTimestamp_datetime()) && getIp().equals(mailMsg.getIp()) && getClientType().equals(mailMsg.getClientType()) && getResult().equals(mailMsg.getResult()) && getVersion().equals(mailMsg.getVersion());
  72.     }
  74.     @Override
  75.     public int hashCode() {
  76.         return Objects.hash(getUser(), getClient_ip(), getSource(), getLoginName(), getMailSenderSourceIp(), getTimestamp_time(), getTimestamp_datetime(), getIp(), getClientType(), getResult(), getVersion());
  77.     }
  79.     @Override
  80.     public String toString() {
  81.         return "{" +
  82.                 "  'user':'" + user + "'" +
  83.                 ", 'client_ip':'" + client_ip  + "'" +
  84.                 ", 'source':'" + source  + "'" +
  85.                 ", 'loginName':'" + loginName  + "'" +
  86.                 ", 'IP':'" + mailSenderSourceIp + "'" +
  87.                 ", 'timestamp':'" + timestamp_time + "'" +
  88.                 ", '@timestamp':'" + timestamp_datetime + "'" +
  89.                 ", 'ip':'"  + "'" +
  90.                 ", 'clientType':'" + clientType  + "'" +
  91.                 ", 'result':'" + result  + "'" +
  92.                 ", 'version':'" + version + "'" +
  93.                 "}";
  94.     }
  96. }

以上就是基于FLink如何实现实时安全检测的详细内容，更多关于基于FLink如何实现实时安全检测的资料请关注九品源码其它相关文章！

相关热词： 基于FLink实现实时安全检测
上一篇：讲一讲如何在一个页面中使用多个百度分享按钮，分享不同的文章
下一篇：Go语言并发编程必备的Mutex互斥锁怎么用

文章分类

IT资讯 CMS教程 服务器 数据库 前端开发 后端开发 工具使用 其他教程

最后更新

热门排行

网站首页 | 关于我们 | 免责声明 | 网站建设 | 在线工具 | 网站地图

版权声明：本站发布的资源部分自互联网，所有资源版权归原作者所有。如果有侵犯到您的权益，请联系本站删除，谢谢合作！

Copyright 2018-2025 ©19jp.com All Rights Reserved. 陕ICP备16002142号-1