028_尚硅谷大数据技术_用户行为数据分析Flink项目_恶意登录检测（二）_简单代码实现

接下来我们的任务就是实现这个自定义的key process function key的process function，因为做过key，所以大家会发现。呃，这里其实。我们所有定义的状态，还有定时器都只针对当前的key有效，当前的user ID有效啊，那所以别的那个不同的用户之间，他们彼此是不被影响的，所以public static class把这个定义出来，Logging field detect warning。那接下来我们既然是process function嘛，啊，那本来它是一个抽象类啊，所以是extends，然后key的process function k process function，它要定义的泛型KIO啊，那当前我们的K的类型是什么呢？啊，这里边KY我们用了这个方法引用，所以user ID是什么类型它就是什么类型，对吧？所以这里边是长整型long，然后接下来输入的数据类型是不是logging event啊。

Logging event输出的数据类型是不是我们包装好的？哎，这里边其实不应该是object，我们想要的是什么？应该是logging，不是包装好了吗？Feel warning对不对，对吧？就是这个类啊，所以接下来这儿的输出类型是不是就是logging feel warning啊，所以大家看这样一写，上面的类型就不报错了，对吧？啊，所以这个看起来还会更加的舒服一点，就是前面我们读取数据读进来之后，诶，这个类型叫做logging event，然后最后我们打印输出啊，或者说写入到外部系统想要的那个数据类型叫做logging warning，然后接下来呢，中间我们就直接分组，然后来一个process，直接放一个大招，这个大招数据类型的转变就是输入logging event，输出logging field war，对吧？啊，非常的直观啊，输入到输出的一个转变，好，那接下来就在这个里面，我们就要就要看定义什么东西了，另外就是这里其实我还可以直接给一个参数。

大家想我当前不是要求连续登录失败两次吗？那大家想如果要是扩展的话，我现在需求变了，变成连续登录失败三次了，五次十次，那是不是可以直接传一个参数进来啊，啊所以这里边我传一个二进来，那首先这应该要定义一个属性对吧。呃，这个属性是连续对登录失败，呃次数就是最多的那个连续登录失败次数，对吧，最大连续登录失败次数，所以这里边我可以定一个private，呃，Integer。给一个整形的次数，比方说我这个就叫max few times对吧？呃，然后接下来。自然是要把这个构造方法要创建出来啊，构造器定义在这儿上面，这个就没问题了，接下来我们要定义的就是。

要定义状态了。我们现在的基本思想不就是要定义一个列表状态，然后把所有当前的这个是登录失败的事件全存进去吗？然后最后等到定时器发生的时候去把这个列表里边的数据，其实不用呃不用都拿出来，大家知道直接就是转成一个列表，取出来的长度是不是就可以了，对吧？只要知道它有有几个数就够了，那判断那个数跟这个max f time相比是不是超过了，如果超过报警，没超过的话就就正常，对吧？清空状态就完事了啊，所以这是我们当前的一个想法，所以定义的状态就是要保存。诶，当前这个两秒内所有的失败事件，登录失败事件。所以我们定义一个list state。里边的类型，我就直接原封不动的把logging event失败的那个事件存进来，对吧，所以它是logging event类型。

我把这个定义成叫log logging，这个其实不叫logging event list state，对吧，应该叫logging feel。List state。我把这个先定义出来，然后另外我还应该要定义一个状态，大家回忆一下，我们之前那个温度连续上升，是不是中间有可能要删定时器啊，删定时器的话，是不是你必须得知道当时注册的时候，它那个时间戳是什么才行，但是注册和我们删不是一个，不是一个时间点，对吧？哎，我并不知道当时是什么时候注册的，所以是不是得有一个状态保存起来啊，而且我们是一个k set不同的，呃，Uz ID，我注册的那个定时器是不是应该是不一样的呀，各自注册各自的，对吧？所以我们还得定义一个状态保存。注册的定时器时间戳，那这个大家知道自然是一个value date了，里边的类型是长整型时间戳嘛，所以这个我们叫做timer ts state，好，这就是基本的状态定义，那当然接下来我们需要在open生命周期里边。

Get当前的运行时环境，然后去做一个定义，对吧，Logging feel list state等于get runtime contact，然后是get list state里边是不是要传一个list，你有一个list state script，对吧？啊，当前的一个描述器，然后里边给一个名称，当前这个我就叫logging feel对吧？List，然后类型，这里边的类型是不是只传里边的具体的那个类型就可以了，所以这里边我给的是logging。event.class好，那初始值的话，我这里面不需要，对吧，初始啥都没有的时候弄就完了嘛，然后同样下边这个timer ts state也要去做一个定义，Get runtime contact get state，这里边是new，一个value state script里边给一个名称，刚才这个叫timer ts，然后当前是长整型的类型，Class放进去，同样初始的时候没有值，那就是none，对吧？啊，这就是我们一开始对于这个所有状态的声明和定义啊，那大家看到现在这个还报错呢，因为里边必须要去重写一个。

对，实现这样一个process element方法，这个才行，对吧，这就是每来一个元素之后，我们接下来要做什么样的一个处理，那大家想一下，现在我们要干什么事呢？其实非常简单，是不是主要就是要判断当前数据，也就是说登录事件的类型啊。大家想想是不是这样，呃，就好比说我们一开始那个，呃，连续温度上升的时候，是不是要判断当前的这个温度值比之前大还是小啊，那现在我们就是判断这个当前事件类型就可以了，所以if。

如果是，如果是fill的话，fill.equals当前的value.get。Logging state对吧。那大家想一下，如果是失败的话怎么办？对，就是如果是失败事件，那是不是，呃，就是要添加，添加到列表状态中，首先是把它先加进去，这个没什么好说的啊，只要是失败了，直接at这个Y6，然后另外至于是不是要注册定时器，那是不是我们还得判断一下有没有定时器啊，对吧？所以这里边是如果没有定时器。那么注册一个两秒之后的定时器，然后所以接下来又一个一服判断有没有定时器，那这个非常简单，是不是我直接用那个t ts state.value看它等不等于弄就可以了呀，对吧，我们现在是没有的话才做操作，有的话就不用管了，对吧？等着定时机触发就完了，所以是判断它等于no的话，接下来要注册了，那我定义一个想要注册的这个时间戳，那就是以当前value里边的time Sam为基准，然后我是不是要加上两秒钟啊。

它本身就是秒，但是我们最后。本身是秒，所以我加二就行了，但是我们最后要的是不是应该是一个毫秒作为单位的一个时间戳啊，所以后面再乘以1000对吧？这就是我们当前要注册的这个当前要注册的这个事件时间的是呃，这个定时器时间戳对吧？所以接下来是不是用c TX time service注册一个事件时间定时器even time timer对吧？啊，所以这里边我们是这样去做的。

把这个TS传进来，另外大家注意是不是还得保存状态啊，既然前面我们有这个嘛，就是你这个地方，如果要是把它update保存了之后，下一次来了之后，如果还是fail的话，一看这里面已经有值了，是不是就不用再去重复注册了啊，所以它是这样的一个原理啊，呃，这是关于如果是失败事件的话，那接下来if这个判断是不是就还有else啊，我们现在事件类型只有两种，那我就直接不用LC if了，直接else就可以。接下来我们上面这个是第一个第一个分支对吧。这是第二个分支，那如果是成功的话。登录成功那怎么办，是不是直接。啊，就删除定时器，然后清空状态重新开始对吧？诶这个就其实非常简单啊，啊，那所以这里边删除定时器呢，还要有一个考虑，那大家想是不是这里面有可能它是能啊。

你如果是nu的话，直接delete一个nu，那那肯定就报错了，对吧，控制人异常了，所以我们还是要确认一下，就是如果有定时器的话才删，没有的话就就。呃，不要去再去操作了，对吧？呃，就是假如说大家想想有没有可能没有没有这个定时器呢？是不是就对，就是我一上来之后就是一个登录成功，那是不是就相当于没有定时器啊，对吧，也没有失败过嘛，那这个时候你就，呃，我们这里边本来没有状态，你清空的话，那当然也没什么事儿，但是你如果要是删定时器的话，那那就报错了，对吧？哎，所以这里边我们要判断一下，如果当前的那个timer ts.value啊，点value，如果说哎，大家想这不是要判断等于了，是不等于的时候我们才清空对不对，不等于none的话，这个时候清空定时器type service现在是deletetime timer，注意这里边给的值就应该是。

对，Timer ts state里边的value对吧？从状态里边获取啊，你可不能拿上面的TS对吧，因为这里边也拿不到前面那个作用域，根本根本是到不了下面这个分支的啊，啊，所以是这样的一个删除的过程，那另外还有清空状态的话，Logging feel event list state直接clear，下面还有一个timer ts state直接clear，这就是完整的流程对吧？啊，这是每一个数据来了之后要做的操作，大家看这个流程我们就都已经搞定了。啊，但是还没完，后边是不是还应该有一个on timer啊，因为你既然注册了定时器嘛，大家想一下现在这个on timer触发的时候会做一个什么事情呢？定时器触发。

定时器触发，诶，那说明至少是两秒内没有登录成功事件来，对不对，就像我们之前那个温度连续上升一样，至少说明它是没有温度下降的那个数据来对吧，但是大家想这里边这个到底是失败了几次，是不是它没准啊，因为你想前面如果我只是登录失败一次，是不是也会注册一个定时器，然后就等着了，哎，所以这里边我们还是得判断它里面的个数，对吧，所以我们要。判断list state中失败的个数，那既然要判断个数，首先是不是我应该要把它这个当前这个list先拿到啊，对吧？哎，所以我还是先转换一下吧，大家还记得我们之前那个lists直接可以newra list里边是不是可以传一个那个interable类型啊，哎，所以我就直接logging fair event list state它去直接get，不是可以得到一个一套过类型吗？对吧？啊，所以我这里边直接可以可以把它传过来啊，直接当前这个我就叫做。

我不要叫logging events，我这个应该叫logging field events对吧，全是登录失败的这个事件，哎，那么大家想我我还是干干脆单独把它取一下吧，啊，就直接把这个对应的这个个数直接拿出来啊，定义一个integer，就当前的这个fair time。Few times，那就是logging feel events里边直接去取他的size，对吧，直接拿到好，那接下来if，就当前的这个few times，如果要大于等于我们前面定义的那个max few times的话，那接下来是不是就得报警啊，对吧？所以这里边就是如果超出设定的。

呃，最最大失败次数，那么输出报警，那大家知道这个输出报警是不是out点。我们这里边报警就是主流，对吧？当前的那个process function不是输出类型，就是那个warning类型吗？啊，所以接下来我们直接包装好out.click把它输出当前这个叫做logging field warning里边我们想要的是当前的用户ID，然后是第一次登录失败的时间，最后一次登录失败的时间，另外还有一个信息，对吧，字符串类型的一个报警信息，大家想当前的这个user ID到哪去拿呢？啊，当然大家会想到这里边这个所有的数据里边是不是都有它user ID啊，或者也不用那么麻烦，大家想对当前ctx里边是不是有key啊，所以当前current key是不是就应该是一个长整型的UID啊，哎，所以我们就是按照uz ID做分组的嘛，所以你get current key就可以了，然后接下来第一次登录失败的那个时间戳怎么拿呢？

那大家想，大家想是不是就是logging few events这个这个a release啊，这个列表里边我是不是直接GET0第一个数啊，然后再拿它的time step是不是就可以了，这就是第一次登录失败的时间戳对吧？好，这个因为数据有点多，我直接换行换过来啊，然后接下来后边还应该有最后一次，那大家想这个类似这个怎么拿get对，我是不是直接用当前的，呃，其实就是few times，前面我不是已经把这个size拿出来了吗？Few times，然后减一是不是就可以了，这是不是最后一个啊，因为从零开始嘛，大家想从零开始，最后一个是不是减一啊，对吧？所以接下来它是get time Sam对吧？同样啊，我把这个也换一行放在这儿。好，那最后是不是再来一个这个message就可以了，我们最后要的就是一个输出的一个报警信息，对吧，比方说我把这个叫做呃logging file在两秒之内，两秒这个我们是写死的对吧，因为这个设置的那个定时器时间戳就是加两秒嘛，啊那这里边我们就是在两秒钟之内。

然后呃，我们直接加一个多少次对吧，For加上当前的few times这里边这么多次对吧？啊。这就是我们最终的一个输出，那当然了，这个大家要注意，定时器既然已经出发了，那该做的那个收尾操作，现在是不是也应该做一个收尾啊，清空状态对吧？清空状态，因为既然定时器已经已经到时间了，那相当于我们这里边这一个周期判断的所有数据都已经搞定了，所以我们直接把这个状态清空就完事了，呃，Logging feel。List state对吧，Event list state做一个clear，然后另外还有timer ts state做一个clear，当然可能有同学想到这个，呃，就直接去把这个完全清空，是不是不对呢？就是大家可能会想到我是不是还应该在，就是把第一个数据他注册的那定时器当然是没有问题了，对吧？那我是不是还应该后面如果要是有有那个，呃，就是比方说后续的那个登录失败的事件，是不是还应该重新注册定时器考察这个两秒钟之内，是不是这个有有连连续两次以上登陆失败呢？啊那那这个就是大家对于逻辑上的一个更加严谨的考虑了，对吧，我们这里边就不考虑那么多了，直接清空，那接下来我们就来运行一下，看看效果怎么样。

跑一下这个代码试试啊。好，我们看一下，诶这边代码输出，这里边已经输出了一个结果了，大家看检测到确实有一条报警，它的显示是1035，这个用户在842和844之间，Logging feel在两秒钟之内三次登录失败，那我们找一下这个。

找一下这个数据啊。诶，大家看到这里边是不是真的是有这个1035啊。1035，而且大家看呃，这这个还是换了IP了，对吧，IP还不一样啊，然后三次连续确实是三次，然后是842843844，确实我们就检测到了，这里面是不是连续三次登录失败啊，啊超过了我们设置的那个两次的登录上限，所以这里边就做了一个输出报警啊，这就是我们做了一个简单的实现。