result．print（＂惡意用戶＞＞＞＂）
   env．execute（＂BarrageBehavior01＂）
 ｝
｝

實(shí)例二：監(jiān)測(cè)刷屏用戶

規(guī)則：用戶如果在10s內(nèi)，同時(shí)連續(xù)輸入同樣一句話超過(guò)5次，就認(rèn)為是惡意刷屏。

使用 Flink CEP檢測(cè)刷屏用戶

object BarrageBehavior02 ｛
 case class Message（userId： String， ip： String， msg： String）
 def main（args： Array［String］）： Unit ＝ ｛
   ／／初始化運(yùn)行環(huán)境
   val env ＝ StreamExecutionEnvironment．getExecutionEnvironment
   ／／設(shè)置并行度
   env．setParallelism（1）
   ／／ 模擬數(shù)據(jù)源
   val loginEventStream： DataStream［Message］ ＝ env．fromCollection（
     List（
       Message（＂1＂， ＂192．168．0．1＂， ＂beijing＂），
       Message（＂1＂， ＂192．168．0．2＂， ＂beijing＂），
       Message（＂1＂， ＂192．168．0．3＂， ＂beijing＂），
       Message（＂1＂， ＂192．168．0．4＂， ＂beijing＂），
       Message（＂2＂， ＂192．168．10．10＂， ＂shanghai＂），
       Message（＂3＂， ＂192．168．10．10＂， ＂beijing＂），
       Message（＂3＂， ＂192．168．10．11＂， ＂beijing＂），
       Message（＂4＂， ＂192．168．10．10＂， ＂beijing＂），
       Message（＂5＂， ＂192．168．10．11＂， ＂shanghai＂），
       Message（＂4＂， ＂192．168．10．12＂， ＂beijing＂），
       Message（＂5＂， ＂192．168．10．13＂， ＂shanghai＂），
       Message（＂5＂， ＂192．168．10．14＂， ＂shanghai＂），
       Message（＂5＂， ＂192．168．10．15＂， ＂beijing＂），
       Message（＂6＂， ＂192．168．10．16＂， ＂beijing＂），
       Message（＂6＂， ＂192．168．10．17＂， ＂beijing＂），
       Message（＂6＂， ＂192．168．10．18＂， ＂beijing＂），
       Message（＂5＂， ＂192．168．10．18＂， ＂shanghai＂），
       Message（＂6＂， ＂192．168．10．19＂， ＂beijing＂），
       Message（＂6＂， ＂192．168．10．19＂， ＂beijing＂），
       Message（＂5＂， ＂192．168．10．18＂， ＂shanghai＂）
     ）
   ）
   ／／定義模式
   val loginbeijingPattern ＝ Pattern．begin［Message］（＂start＂）
     ．where（＿．msg �。� null） ／／一條登錄失敗
     ．times（5）．optional  ／／將滿足五次的數(shù)據(jù)配對(duì)打印
     ．within（Time．seconds（10））
   ／／進(jìn)行分組匹配
   val loginbeijingDataPattern ＝ CEP．pattern（loginEventStream．keyBy（＿．userId）， loginbeijingPattern）
   ／／查找符合規(guī)則的數(shù)據(jù)
   val loginbeijingResult： DataStream［Option［Iterable［Message］］］ ＝ loginbeijingDataPattern．select（patternSelectFun ＝ （pattern： collection．Map［String， Iterable［Message］］） ＝＞ ｛
     var loginEventList： Option［Iterable［Message］］ ＝ null
     loginEventList ＝ pattern．get（＂start＂） match ｛
       case Some（value） ＝＞ ｛
         if （value．toList．map（x ＝＞ （x．userId， x．msg））．distinct．size ＝＝ 1） ｛
           Some（value）
         ｝ else ｛
           None
         ｝
       ｝
     ｝
     loginEventList
   ｝）
   ／／打印測(cè)試
   loginbeijingResult．filter（x＝＞x�。絅one）．map（x＝＞｛
     x match ｛
       case Some（value）＝＞ value
     ｝
   ｝）．print（）
   env．execute（＂BarrageBehavior02）
 ｝
｝
Flink CEP API

除了案例中介紹的幾個(gè)API外，我們?cè)诮榻B下其他的常用API：

1． 條件 API

為了讓傳入事件被模式所接受，給模式指定傳入事件必須滿足的條件，這些條件由事件本身的屬性或者前面匹配過(guò)的事件的屬性統(tǒng)計(jì)量等來(lái)設(shè)定。比如，事件的某個(gè)值大于5，或者大于先前接受事件的某個(gè)值的平均值。

可以使用pattern．where（）、pattern．or（）、pattern．until（）方法來(lái)指定條件。條件既可以是迭代條件IterativeConditions，也可以是簡(jiǎn)單條件SimpleConditions。

FlinkCEP支持事件之間的三種臨近條件：

next（）：嚴(yán)格的滿足條件

示例：模式為begin（＂first＂）．where（＿．name＝＇a＇）．next（＂second＂）．where（．name＝＇b＇）當(dāng)且僅當(dāng)數(shù)據(jù)為a，b時(shí)，模式才會(huì)被命中。如果數(shù)據(jù)為a，c，b，由于a的后面跟了c，所以a會(huì)被直接丟棄，模式不會(huì)命中。

followedBy（）：松散的滿足條件

示例：模式為begin（＂first＂）．where（＿．name＝＇a＇）．followedBy（＂second＂）．where（．name＝＇b＇）當(dāng)且僅當(dāng)數(shù)據(jù)為a，b或者為a，c，b，模式均被命中，中間的c會(huì)被忽略掉。

followedByAny（）：非確定的松散滿足條件

示例：模式為begin（＂first＂）．where（＿．name＝＇a＇）．followedByAny（＂second＂）．where（．name＝＇b＇）當(dāng)且僅當(dāng)數(shù)據(jù)為a，c，b，b時(shí)，對(duì)于followedBy模式而言命中的為｛a，b｝，對(duì)于followedByAny而言會(huì)有兩次命中｛a，b｝，｛a，b｝。

2． 量詞 API

還記得我們?cè)谏厦嬷v解模式概念時(shí)說(shuō)過(guò)的一句話：一般情況下，模式都是單例模式，可以使用量詞（Quantifiers）將其轉(zhuǎn)換為循環(huán)模式。這里的量詞就是指的量詞API。

以下這幾個(gè)量詞API，可以將模式指定為循環(huán)模式：

pattern．oneOrMore（）：一個(gè)給定的事件有一次或多次出現(xiàn)，例如上面提到的b＋。

pattern．times（＃ofTimes）：一個(gè)給定類型的事件出現(xiàn)了指定次數(shù)，例如4次。

pattern．times（＃fromTimes， ＃toTimes）：一個(gè)給定類型的事件出現(xiàn)的次數(shù)在指定次數(shù)范圍內(nèi)，例如2～4次。

可以使用pattern．greedy（）方法將模式變成循環(huán)模式，但是不能讓一組模式都變成循環(huán)模式。greedy：就是盡可能的重復(fù)。

使用pattern．optional（）方法將循環(huán)模式變成可選的，即可以是循環(huán)模式也可以是單個(gè)模式。

3． 匹配后的跳過(guò)策略

所謂的匹配跳過(guò)策略，是對(duì)多個(gè)成功匹配的模式進(jìn)行篩選。也就是說(shuō)如果多個(gè)匹配成功，可能我不需要這么多，按照匹配策略，過(guò)濾下就可以。

Flink中有五種跳過(guò)策略：

NO＿SKIP： 不過(guò)濾，所有可能的匹配都會(huì)被發(fā)出。

SKIP＿TO＿NEXT： 丟棄與開(kāi)始匹配到的事件相同的事件，發(fā)出開(kāi)始匹配到的事件，即直接跳到下一個(gè)模式匹配到的事件，以此類推。

SKIP＿PAST＿LAST＿EVENT： 丟棄匹配開(kāi)始后但結(jié)束之前匹配到的事件。

SKIP＿TO＿FIRST［PatternName］： 丟棄匹配開(kāi)始后但在PatternName模式匹配到的第一個(gè)事件之前匹配到的事件。

SKIP＿TO＿LAST［PatternName］： 丟棄匹配開(kāi)始后但在PatternName模式匹配到的最后一個(gè)事件之前匹配到的事件。

怎么理解上述策略，我們以NO＿SKIP和SKIP＿PAST＿LAST＿EVENT為例講解下：

在模式為：begin（＂start＂）．where（＿．name＝＇a＇）．oneOrMore（）．followedBy（＂second＂）．where（＿．name＝＇b＇）中，我們輸入數(shù)據(jù)：a，a，a，a，b ，如果是NO＿SKIP策略，即不過(guò)濾策略，模式匹配到的是：｛a，b｝，｛a，a，b｝，｛a，a，a，b｝，｛a，a，a，a，b｝；如果是SKIP＿PAST＿LAST＿EVENT策略，即丟棄匹配開(kāi)始后但結(jié)束之前匹配到的事件，模式匹配到的是：｛a，a，a，a，b｝。

Flink CEP 的使用場(chǎng)景

除上述案例場(chǎng)景外，F(xiàn)link CEP 還廣泛用于網(wǎng)絡(luò)欺詐，故障檢測(cè)，風(fēng)險(xiǎn)規(guī)避，智能營(yíng)銷等領(lǐng)域。

1． 實(shí)時(shí)反作弊和風(fēng)控

對(duì)于電商來(lái)說(shuō)，羊毛黨是必不可少的，國(guó)內(nèi)拼多多曾爆出 100 元的無(wú)門檻券隨便領(lǐng)，當(dāng)晚被人褥幾百億，對(duì)于這種情況肯定是沒(méi)有做好及時(shí)的風(fēng)控。另外還有就是商家上架商品時(shí)通過(guò)頻繁修改商品的名稱和濫用標(biāo)題來(lái)提高搜索關(guān)鍵字的排名、批量注冊(cè)一批機(jī)器賬號(hào)快速刷單來(lái)提高商品的銷售量等作弊行為，各種各樣的作弊手法也是需要不斷的去制定規(guī)則去匹配這種行為。

2． 實(shí)時(shí)營(yíng)銷

分析用戶在手機(jī) APP 的實(shí)時(shí)行為，統(tǒng)計(jì)用戶的活動(dòng)周期，通過(guò)為用戶畫像來(lái)給用戶進(jìn)行推薦。比如用戶在登錄 APP 后 1 分鐘內(nèi)只瀏覽了商品沒(méi)有下單；用戶在瀏覽一個(gè)商品后，3 分鐘內(nèi)又去查看其他同類的商品，進(jìn)行比價(jià)行為；用戶商品下單后 1 分鐘內(nèi)是否支付了該訂單。如果這些數(shù)據(jù)都可以很好的利用起來(lái)，那么就可以給用戶推薦瀏覽過(guò)的類似商品，這樣可以大大提高購(gòu)買率。

3． 實(shí)時(shí)網(wǎng)絡(luò)攻擊檢測(cè)

當(dāng)下互聯(lián)網(wǎng)安全形勢(shì)仍然嚴(yán)峻，網(wǎng)絡(luò)攻擊屢見(jiàn)不鮮且花樣眾多，這里我們以 DDOS（分布式拒絕服務(wù)攻擊）產(chǎn)生的流入流量來(lái)作為遭受攻擊的判斷依據(jù)。對(duì)網(wǎng)絡(luò)遭受的潛在攻擊進(jìn)行實(shí)時(shí)檢測(cè)并給出預(yù)警，云服務(wù)廠商的多個(gè)數(shù)據(jù)中心會(huì)定時(shí)向監(jiān)控中心上報(bào)其瞬時(shí)流量，如果流量在預(yù)設(shè)的正常范圍內(nèi)則認(rèn)為是正�，F(xiàn)象，不做任何操作；如果某數(shù)據(jù)中心在 10 秒內(nèi)連續(xù) 5 次上報(bào)的流量超過(guò)正常范圍的閾值，則觸發(fā)一條警告的事件；如果某數(shù)據(jù)中心 30 秒內(nèi)連續(xù)出現(xiàn) 30 次上報(bào)的流量超過(guò)正常范圍的閾值，則觸發(fā)嚴(yán)重的告警。

Flink CEP 的原理簡(jiǎn)單介紹

Apache Flink在實(shí)現(xiàn)CEP時(shí)借鑒了Efficient Pattern Matching over Event Streams論文中NFA的模型，在這篇論文中，還提到了一些優(yōu)化，我們?cè)谶@里先跳過(guò)，只說(shuō)下NFA的概念。

在這篇論文中，提到了NFA，也就是Non－determined Finite Automaton，叫做不確定的有限狀態(tài)機(jī)，指的是狀態(tài)有限，但是每個(gè)狀態(tài)可能被轉(zhuǎn)換成多個(gè)狀態(tài)（不確定）。

非確定有限自動(dòng)狀態(tài)機(jī)：

先介紹兩個(gè)概念：

狀態(tài)：狀態(tài)分為三類，起始狀態(tài)、中間狀態(tài)和最終狀態(tài)。

轉(zhuǎn)換：take／ignore／proceed都是轉(zhuǎn)換的名稱。

在NFA匹配規(guī)則里，本質(zhì)上是一個(gè)狀態(tài)轉(zhuǎn)換的過(guò)程。三種轉(zhuǎn)換的含義如下所示：

Take： 主要是條件的判斷，當(dāng)過(guò)來(lái)一條數(shù)據(jù)進(jìn)行判斷，一旦滿足條件，獲取當(dāng)前元素，放入到結(jié)果集中，然后將當(dāng)前狀態(tài)轉(zhuǎn)移到下一個(gè)的狀態(tài)。

Proceed：當(dāng)前的狀態(tài)可以不依賴任何的事件轉(zhuǎn)移到下一個(gè)狀態(tài)，比如說(shuō)透?jìng)鞯囊馑肌?/p>

Ignore：當(dāng)一條數(shù)據(jù)到來(lái)的時(shí)候，可以忽略這個(gè)消息事件，當(dāng)前的狀態(tài)保持不變，相當(dāng)于自己到自己的一個(gè)狀態(tài)。

NFA的特點(diǎn)：在NFA中，給定當(dāng)前狀態(tài)，可能有多個(gè)下一個(gè)狀態(tài)。可以隨機(jī)選擇下一個(gè)狀態(tài)，也可以并行（同時(shí)）選擇下一個(gè)狀態(tài)。輸入符號(hào)可以為空。

規(guī)則引擎

規(guī)則引擎：將業(yè)務(wù)決策從應(yīng)用程序代碼中分離出來(lái)，并使用預(yù)定義的語(yǔ)義模塊編寫業(yè)務(wù)決策。接受數(shù)據(jù)輸入，解釋業(yè)務(wù)規(guī)則，并根據(jù)業(yè)務(wù)規(guī)則做出業(yè)務(wù)決策。
使用規(guī)則引擎可以通過(guò)降低實(shí)現(xiàn)復(fù)雜業(yè)務(wù)邏輯的組件的復(fù)雜性，降低應(yīng)用程序的維護(hù)和可擴(kuò)展性成本。

1． Drools

Drools 是一款使用 Java 編寫的開(kāi)源規(guī)則引擎，通常用來(lái)解決業(yè)務(wù)代碼與業(yè)務(wù)規(guī)則的分離，它內(nèi)置的 Drools Fusion 模塊也提供 CEP 的功能。

優(yōu)勢(shì)：

功能較為完善，具有如系統(tǒng)監(jiān)控、操作平臺(tái)等功能。規(guī)則支持動(dòng)態(tài)更新。

劣勢(shì)：

以內(nèi)存實(shí)現(xiàn)時(shí)間窗功能，無(wú)法支持較長(zhǎng)跨度的時(shí)間窗。無(wú)法有效支持定時(shí)觸達(dá)（如用戶在瀏覽發(fā)生一段時(shí)間后觸達(dá)條件判斷）。2． Aviator

Aviator 是一個(gè)高性能、輕量級(jí)的 Java 語(yǔ)言實(shí)現(xiàn)的表達(dá)式求值引擎，主要用于各種表達(dá)式的動(dòng)態(tài)求值。

優(yōu)勢(shì)：

支持大部分運(yùn)算操作符。支持函數(shù)調(diào)用和自定義函數(shù)。支持正則表達(dá)式匹配。支持傳入變量并且性能優(yōu)秀。

劣勢(shì)：

沒(méi)有 if else、do while 等語(yǔ)句，沒(méi)有賦值語(yǔ)句，沒(méi)有位運(yùn)算符。3． EasyRules

EasyRules 集成了 MVEL 和 SpEL 表達(dá)式的一款輕量級(jí)規(guī)則引擎。

優(yōu)勢(shì)：

輕量級(jí)框架，學(xué)習(xí)成本低。基于 POJO。為定義業(yè)務(wù)引擎提供有用的抽象和簡(jiǎn)便的應(yīng)用。支持從簡(jiǎn)單的規(guī)則組建成復(fù)雜規(guī)則。4． Esper

Esper 設(shè)計(jì)目標(biāo)為 CEP 的輕量級(jí)解決方案，可以方便的嵌入服務(wù)中，提供 CEP 功能。

優(yōu)勢(shì)：

輕量級(jí)可嵌入開(kāi)發(fā)，常用的 CEP 功能簡(jiǎn)單好用。EPL 語(yǔ)法與 SQL 類似，學(xué)習(xí)成本較低。

劣勢(shì)：

單機(jī)全內(nèi)存方案，需要整合其他分布式和存儲(chǔ)。以內(nèi)存實(shí)現(xiàn)時(shí)間窗功能，無(wú)法支持較長(zhǎng)跨度的時(shí)間窗。無(wú)法有效支持定時(shí)觸達(dá)（如用戶在瀏覽發(fā)生一段時(shí)間后觸達(dá)條件判斷）。5． Flink CEP

Flink 是一個(gè)流式系統(tǒng)，具有高吞吐低延遲的特點(diǎn)，F(xiàn)link CEP 是一套極具通用性、易于使用的實(shí)時(shí)流式事件處理方案。

優(yōu)勢(shì)：

繼承了 Flink 高吞吐的特點(diǎn)。事件支持存儲(chǔ)到外部，可以支持較長(zhǎng)跨度的時(shí)間窗�？梢灾С侄〞r(shí)觸達(dá)（用 followedBy ＋ PartternTimeoutFunction 實(shí)現(xiàn)）。