概要

        實(shí)時(shí)風(fēng)控解決方案
        總體架構(gòu)介紹
        第一版需求實(shí)現(xiàn)

1. 風(fēng)控背景

      互聯(lián)網(wǎng)場(chǎng)景中，典型的風(fēng)控場(chǎng)景包括：注冊(cè)風(fēng)控、登陸風(fēng)控、交易風(fēng)控、活動(dòng)風(fēng)控等，而風(fēng)控的最佳效果是防患于未然，所以事前事中和事后三種實(shí)現(xiàn)方案中，又以事前預(yù)警和事中控制最好。
      這要求風(fēng)控系統(tǒng)一定要有實(shí)時(shí)性。我們將實(shí)時(shí)風(fēng)控架構(gòu)作為重點(diǎn)講解。

2. 總體架構(gòu)

       風(fēng)控是業(yè)務(wù)場(chǎng)景的產(chǎn)物，風(fēng)控系統(tǒng)直接服務(wù)于業(yè)務(wù)系統(tǒng)，與之相關(guān)的還有懲罰系統(tǒng)和分析系統(tǒng)，各系統(tǒng)關(guān)系與角色如下：

風(fēng)控系統(tǒng)有規(guī)則和模型兩種技術(shù)路線，規(guī)則的優(yōu)點(diǎn)是簡(jiǎn)單直觀、可解釋性強(qiáng)、靈活，所以長(zhǎng)期活躍在風(fēng)控系統(tǒng)之中，但缺點(diǎn)是容易被攻破，一但被黑產(chǎn)猜中就會(huì)失效，于是在實(shí)際的風(fēng)控系統(tǒng)中，往往需要再結(jié)合上基于模型的風(fēng)控環(huán)節(jié)來(lái)增加健壯性。

• 規(guī)則就是針對(duì)事物的條件判斷，我們針對(duì)注冊(cè)、登陸、交易、活動(dòng)分別假設(shè)幾條規(guī)則，比如：

  • 用戶名與身份證姓名不一致；

  • 某 IP 最近 1 小時(shí)注冊(cè)賬號(hào)數(shù)超過 10 個(gè)；

  • 某賬號(hào)最近 3 分鐘登陸次數(shù)大于 5 次；

  • 某賬號(hào)群體最近 1 小時(shí)購(gòu)買優(yōu)惠商品超過 100 件；

  • 某賬號(hào)最近 3 分鐘領(lǐng)券超過 3 張；

• 規(guī)則可以組合成規(guī)則組

  • 事實(shí)：即被判斷的主體和屬性，如上面規(guī)則的賬號(hào)及登陸次數(shù)、IP 和注冊(cè)次數(shù)等；

  • 條件：判斷的邏輯，如某事實(shí)的某屬性大于某個(gè)指標(biāo)；

  • 指標(biāo)閾值：判斷的依據(jù)，比如登陸次數(shù)的臨界閾值，注冊(cè)賬號(hào)數(shù)的臨界閾值等；

  • 規(guī)則可由運(yùn)營(yíng)專家憑經(jīng)驗(yàn)填寫，也可由數(shù)據(jù)分析師根據(jù)歷史數(shù)據(jù)發(fā)掘，但因?yàn)橐?guī)則在與黑產(chǎn)的攻防之中會(huì)被猜中導(dǎo)致失效，所以無(wú)一例外都需要?jiǎng)討B(tài)調(diào)整。

• 基于上邊的討論，我們?cè)O(shè)計(jì)一個(gè)風(fēng)控系統(tǒng)方案如下：

• 該系統(tǒng)有三條數(shù)據(jù)流向：

  • 實(shí)時(shí)風(fēng)控?cái)?shù)據(jù)流：由紅線標(biāo)識(shí)，同步調(diào)用，為風(fēng)控調(diào)用的核心鏈路

  • 準(zhǔn)實(shí)時(shí)指標(biāo)數(shù)據(jù)流：由藍(lán)線標(biāo)識(shí)，異步寫入，為實(shí)時(shí)風(fēng)控部分準(zhǔn)備指標(biāo)數(shù)據(jù)

  • 準(zhǔn)實(shí)時(shí)/離線分析數(shù)據(jù)流：由綠線標(biāo)識(shí)，異步寫入，為風(fēng)控系統(tǒng)的表現(xiàn)分析提供數(shù)據(jù)

實(shí)時(shí)風(fēng)控

        實(shí)時(shí)風(fēng)控是整個(gè)系統(tǒng)的核心，被業(yè)務(wù)系統(tǒng)同步調(diào)用，完成對(duì)應(yīng)的風(fēng)控判斷。
        前面提到規(guī)則往往由人編寫并且需要?jiǎng)討B(tài)調(diào)整，所以我們會(huì)把風(fēng)控判斷部分與規(guī)則管理部分拆開。規(guī)則管理后臺(tái)為運(yùn)營(yíng)服務(wù)，由運(yùn)營(yíng)人員去進(jìn)行相關(guān)操作：
        場(chǎng)景管理：決定某個(gè)場(chǎng)景是否實(shí)施風(fēng)控，比如活動(dòng)場(chǎng)景，在活動(dòng)結(jié)束后可以關(guān)閉該場(chǎng)景；
        黑白名單：人工/程序找到系統(tǒng)的黑白名單，直接過濾
        規(guī)則管理：管理規(guī)則，包括增刪或修改，比如登陸新增 IP 地址判斷，比如下單新增頻率校驗(yàn)等；
        閾值管理：管理指標(biāo)的閾值，比如規(guī)則為某 IP 最近 1 小時(shí)注冊(cè)賬號(hào)數(shù)不能超過 10 個(gè)，那 1 和 10 都屬于閾值；
        基于管理后臺(tái)，那規(guī)則判斷部分的邏輯也就十分清晰了，分別包括前置過濾、事實(shí)數(shù)據(jù)準(zhǔn)備、規(guī)則判斷三個(gè)環(huán)節(jié)。

• 前置過濾

  • 業(yè)務(wù)系統(tǒng)在特定事件（如注冊(cè)、登陸、下單、參加活動(dòng)等）被觸發(fā)后同步調(diào)用風(fēng)控系統(tǒng)，附帶相關(guān)上下文，比如 IP 地址，事件標(biāo)識(shí)等，規(guī)則判斷部分會(huì)根據(jù)管理后臺(tái)的配置決定是否進(jìn)行判斷，如果是，接著進(jìn)行黑白名單過濾，都通過后進(jìn)入下一個(gè)環(huán)節(jié)。

• 實(shí)時(shí)數(shù)據(jù)準(zhǔn)備

  • 在進(jìn)行判斷之前，系統(tǒng)必須要準(zhǔn)備一些事實(shí)數(shù)據(jù)，比如：

    • 注冊(cè)場(chǎng)景，假如規(guī)則為單一 IP 最近 1 小時(shí)注冊(cè)賬號(hào)數(shù)不超過 10 個(gè)，那系統(tǒng)需要根據(jù) IP 地址去 Redis/Hbase 找到該 IP 最近 1 小時(shí)注冊(cè)賬號(hào)的數(shù)目，比如 15；

    • 登陸場(chǎng)景，假如規(guī)則為單一賬號(hào)最近 3 分鐘登陸次數(shù)不超過 5 次，那系統(tǒng)需要根據(jù)賬號(hào)去 Redis/Hbase 找到該賬號(hào)最近 3 分鐘登陸的次數(shù)，比如 8；

規(guī)則判斷

        在得到事實(shí)數(shù)據(jù)之后，系統(tǒng)會(huì)根據(jù)規(guī)則和閾值進(jìn)行判斷，然后返回結(jié)果，整個(gè)過程便結(jié)束了。
        整個(gè)過程邏輯上是清晰的，我們常說的規(guī)則引擎主要在這部分起作用，一般來(lái)說這個(gè)過程有兩種實(shí)現(xiàn)方式：
        借助成熟的規(guī)則引擎，比如 Drools，Drools 和 Java 環(huán)境結(jié)合的非常好，本身也非常完善，支持很多特性。
        基于 Groovy 等動(dòng)態(tài)語(yǔ)言完成
        這兩種方案都支持規(guī)則的動(dòng)態(tài)更新

• 準(zhǔn)實(shí)時(shí)數(shù)據(jù)流

  • 這部分屬于后臺(tái)邏輯，為風(fēng)控系統(tǒng)服務(wù)，準(zhǔn)備事實(shí)數(shù)據(jù)。

  • 把數(shù)據(jù)準(zhǔn)備與邏輯判斷拆分，是出于系統(tǒng)的性能/可擴(kuò)展性的角度考慮的。前邊提到，做規(guī)則判斷需要事實(shí)的相關(guān)指標(biāo)，比如最近一小時(shí)登陸次數(shù)，最近一小時(shí)注冊(cè)賬號(hào)數(shù)等等，這些指標(biāo)通常有一段時(shí)間跨度，是某種狀態(tài)或聚合，很難在實(shí)時(shí)風(fēng)控過程中根據(jù)原始數(shù)據(jù)進(jìn)行計(jì)算，因?yàn)轱L(fēng)控的規(guī)則引擎往往是無(wú)狀態(tài)的，不會(huì)記錄前面的結(jié)果。

  • 同時(shí)，這部分原始數(shù)據(jù)量很大，因?yàn)橛脩艋顒?dòng)的原始數(shù)據(jù)都要傳過來(lái)進(jìn)行計(jì)算，所以這部分往往由一個(gè)流式大數(shù)據(jù)系統(tǒng)來(lái)完成。

  • 業(yè)務(wù)系統(tǒng)把埋點(diǎn)數(shù)據(jù)發(fā)送到 Kafka；

    • Flink 訂閱 Kafka，完成原子粒度的聚合；

  • Flink 僅完成原子粒度的聚合是和規(guī)則的動(dòng)態(tài)變更邏輯相關(guān)的。舉例來(lái)說，在注冊(cè)場(chǎng)景中，運(yùn)營(yíng)同學(xué)會(huì)根據(jù)效果一會(huì)要判斷某 IP 最近 1 小時(shí)的注冊(cè)賬號(hào)數(shù)，一會(huì)要判斷最近 3 小時(shí)的注冊(cè)賬號(hào)數(shù)，一會(huì)又要判斷最近 5 小時(shí)的注冊(cè)賬號(hào)數(shù)……也就是說這個(gè)最近 N 小時(shí)的 N 是動(dòng)態(tài)調(diào)整的。那 Flink 在計(jì)算時(shí)只應(yīng)該計(jì)算 1 小時(shí)的賬號(hào)數(shù)，在判斷過程中根據(jù)規(guī)則來(lái)讀取最近 3 個(gè) 1 小時(shí)還是 5 個(gè) 1 小時(shí)，然后聚合后進(jìn)行判斷。因?yàn)樵?Flink 的運(yùn)行機(jī)制中，作業(yè)提交后會(huì)持續(xù)運(yùn)行，如果調(diào)整邏輯需要停止作業(yè)，修改代碼，然后重啟，相當(dāng)麻煩；同時(shí)因?yàn)?Flink 中間狀態(tài)的問題，重啟還面臨著中間狀態(tài)能否復(fù)用的問題。所以假如直接由 Flink 完成 N 小時(shí)的聚合的話，每次 N 的變動(dòng)都需要重復(fù)上面的操作，有時(shí)還需要追數(shù)據(jù)，非常繁瑣。

    • Flink 把匯總的指標(biāo)結(jié)果寫入 Redis 或 Hbase，供實(shí)時(shí)風(fēng)控系統(tǒng)查詢。兩者問題都不大，根據(jù)場(chǎng)景選擇即可。

    • 通過把數(shù)據(jù)計(jì)算和邏輯判斷拆分開來(lái)并引入 Flink，我們的風(fēng)控系統(tǒng)可以應(yīng)對(duì)極大的用戶規(guī)模。

分析系統(tǒng)
       前面的東西靜態(tài)來(lái)看是一個(gè)完整的風(fēng)控系統(tǒng)，但動(dòng)態(tài)來(lái)看就有缺失了，這種缺失不體現(xiàn)在功能性上，而是體現(xiàn)在演進(jìn)上。即如果從動(dòng)態(tài)的角度來(lái)看一個(gè)風(fēng)控系統(tǒng)的話，我們至少還需要兩部分，一是衡量系統(tǒng)的整體效果，一是為系統(tǒng)提供規(guī)則/邏輯升級(jí)的依據(jù)。
       在衡量整體效果方面，我們需要：
            判斷規(guī)則是否失效，比如攔截率的突然降低；
            判斷規(guī)則是否多余，比如某規(guī)則從來(lái)沒攔截過任何事件；
            判斷規(guī)則是否有漏洞，比如在舉辦某個(gè)促銷活動(dòng)或發(fā)放代金券后，福利被領(lǐng)完了，但沒有達(dá)到預(yù)期效果；
       在為系統(tǒng)提供規(guī)則/邏輯升級(jí)依據(jù)方面，我們需要：
            發(fā)現(xiàn)全局規(guī)則：比如某人在電子產(chǎn)品的花費(fèi)突然增長(zhǎng)了 100 倍，單獨(dú)來(lái)看是有問題的，但整體來(lái)看，可能很多人都出現(xiàn)了這個(gè)現(xiàn)象，原來(lái)是蘋果發(fā)新品了。
           識(shí)別某種行為的組合：?jiǎn)未涡袨槭钦５?，但組合是異常的，比如用戶買菜刀是正常的，買車票是正常的，買繩子也是正常的，去加油站加油也是正常的，但短時(shí)間內(nèi)同時(shí)做這些事情就不是正常的。
            群體識(shí)別：比如通過圖分析技術(shù)，發(fā)現(xiàn)某個(gè)群體，然后給給這個(gè)群體的所有賬號(hào)都打上群體標(biāo)簽，防止出現(xiàn)那種每個(gè)賬號(hào)表現(xiàn)都正常，但整個(gè)群體卻在集中薅羊毛的情況。
            這便是分析系統(tǒng)的角色定位，在他的工作中有部分是確定性的，也有部分是探索性的，為了完成這種工作，該系統(tǒng)需要盡可能多的數(shù)據(jù)支持，如：
            業(yè)務(wù)系統(tǒng)的數(shù)據(jù)，業(yè)務(wù)的埋點(diǎn)數(shù)據(jù)，記錄詳細(xì)的用戶、交易或活動(dòng)數(shù)據(jù)；
            風(fēng)控?cái)r截?cái)?shù)據(jù)，風(fēng)控系統(tǒng)的埋點(diǎn)數(shù)據(jù)，比如某個(gè)用戶在具有某些特征的狀態(tài)下因?yàn)槟硹l規(guī)則而被攔截，這條攔截本身就是一個(gè)事件數(shù)據(jù)；
            這是一個(gè)典型的大數(shù)據(jù)分析場(chǎng)景，架構(gòu)也比較靈活

      相對(duì)來(lái)說這個(gè)系統(tǒng)是最開放的，既有固定的指標(biāo)分析，也可以使用機(jī)器學(xué)習(xí)/數(shù)據(jù)分析技術(shù)發(fā)現(xiàn)更多新的規(guī)則或模式。 

3.第一版需求開發(fā)

數(shù)據(jù)源介紹：

包名：com.star.engine.pojo

日志實(shí)體類

package com.star.engine.pojo;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;
import java.util.Map;
@Data
@NoArgsConstructor
@AllArgsConstructor
@ToString
public class ClientLog {
    private String userNo;  // 用戶ID
    private String userName;  // 用戶名
    private String appId;  // app的編號(hào)
    private String appVersion; // APP的版本
    private String addr;  // 地址
    private String carrier; // 運(yùn)營(yíng)商
    private String imei; // 設(shè)備編號(hào)
    private String deviceType; // 設(shè)備類型
    private String ip;  // 客戶端IP
    private String netType; // 網(wǎng)絡(luò)類型：WIFI,4G,5G
    private String osName; // 操作系統(tǒng)類型
    private String osVersion;  // 操作系統(tǒng)版本
    private String sessionId; // 會(huì)話ID
    private String detailTime; // 創(chuàng)建詳細(xì)時(shí)間
    private String eventId; // 事件編號(hào)
    private String eventType;  // 事件類型
    private String createTime;  // 創(chuàng)建時(shí)間
    private String gps; // 經(jīng)緯度信息
    private Map<String,String> properties; // 事件詳細(xì)屬性
}

• 工具類介紹

包名：com.star.engine.utils;

Constants 常量類

package com.star.engine.utils;
/**
 * 常量信息
 */
public class Constants {
    //topic
    public static String CLIENT_LOG = "client_log";
    public static String HBASE_TABLE = "events_db:users";
    public static String BROKERS = "star01:9092,star02:9092,star03:9092";
    public static Integer REDIS_PORT = 6379;
    public static String HOST = "star01";
    public static String REDIS_ADDR = "star01";
    public static String ZOOKEEPER_PORT = "2181";
    public static String RULE_TYPE_LOGIN = "login";
}

FlinkKafkaUtils FlinkKafka工具類

package com.star.engine.utils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import java.util.Properties;
public class FlinkKafkaUtils {
    public static Properties getProducerProperties(String brokers) {
        Properties properties = getCommonProperties();
        properties.setProperty("bootstrap.servers", brokers);
        properties.setProperty("metadata.broker.list", brokers);
        properties.setProperty("zookeeper.connect", Constants.HOST+":"+Constants.ZOOKEEPER_PORT);
        return properties;
    }
    public static Properties getCommonProperties() {
        Properties properties = new Properties();
        properties.setProperty("linger.ms", "100");
        properties.setProperty("retries", "100");
        properties.setProperty("retry.backoff.ms", "200");
        properties.setProperty("buffer.memory", "524288");
        properties.setProperty("batch.size", "100");
        properties.setProperty("max.request.size", "524288");
        properties.setProperty("compression.type", "snappy");
        properties.setProperty("request.timeout.ms", "180000");
        properties.setProperty("max.block.ms", "180000");
        return properties;
    }
    public static FlinkKafkaConsumer<String> getKafkaEventSource(){
        Properties props = getProducerProperties(Constants.BROKERS);
        props.setProperty("auto.offset.reset", "latest");
        //指定Topic
        FlinkKafkaConsumer<String> source = new FlinkKafkaConsumer<>(Constants.CLIENT_LOG, new SimpleStringSchema(), props);
        return source;
    }
    public static FlinkKafkaConsumer<String> getKafkaRuleSource() {
        Properties props = getProducerProperties(Constants.BROKERS);
        props.setProperty("auto.offset.reset", "latest");
        FlinkKafkaConsumer<String> source = new FlinkKafkaConsumer<>("yinew_drl_rule", new SimpleStringSchema(), props);
        return source;
    }
}

HBaseUtils Hbase工具類

package com.star.engine.utils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
import java.util.List;
public class HBaseUtils {
    private static Connection connection;
    private static Configuration configuration;
    static {
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.property.clientPort", Constants.ZOOKEEPER_PORT);
        configuration.set("hbase.zookeeper.quorum", Constants.HOST);
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public static HTable initHbaseClient(String tableName) {
        try {
            return new HTable(configuration, tableName);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    /**
     * 創(chuàng)建 HBase 表
     * @param tableName      表名
     * @param columnFamilies 列族的數(shù)組
     */
    public static boolean createTable(String tableName, List<String> columnFamilies) {
        try {
            HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
            if (admin.tableExists(tableName)) {
                return false;
            }
            HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
            columnFamilies.forEach(columnFamily -> {
                HColumnDescriptor columnDescriptor = new HColumnDescriptor(columnFamily);
                columnDescriptor.setMaxVersions(2);
                tableDescriptor.addFamily(columnDescriptor);
            });
            admin.createTable(tableDescriptor);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return true;
    }
    /**
     * 刪除 hBase 表
     * @param tableName 表名
     */
    public static boolean deleteTable(String tableName) {
        try {
            HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
            // 刪除表前需要先禁用表
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }
    /**
     * 插入數(shù)據(jù)
     *
     * @param tableName        表名
     * @param rowKey           唯一標(biāo)識(shí)
     * @param columnFamilyName 列族名
     * @param qualifier        列標(biāo)識(shí)
     * @param value            數(shù)據(jù)
     */
    public static boolean putRow(String tableName, String rowKey, String columnFamilyName, String qualifier,
                                 String value) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(qualifier), Bytes.toBytes(value));
            table.put(put);
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return true;
    }
    /**
     * 根據(jù) rowKey 獲取指定行的數(shù)據(jù)
     *
     * @param tableName 表名
     * @param rowKey    唯一標(biāo)識(shí)
     */
    public static Result getRow(String tableName, String rowKey) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            return table.get(get);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    /**
     * 獲取指定行指定列 (cell) 的最新版本的數(shù)據(jù)
     *
     * @param tableName    表名
     * @param rowKey       唯一標(biāo)識(shí)
     * @param columnFamily 列族
     * @param qualifier    列標(biāo)識(shí)
     */
    public static String getCell(String tableName, String rowKey, String columnFamily, String qualifier) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            if (!get.isCheckExistenceOnly()) {
                get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
                Result result = table.get(get);
                byte[] resultValue = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
                return Bytes.toString(resultValue);
            } else {
                return null;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    /**
     * 檢索全表
     *
     * @param tableName 表名
     */
    public static ResultScanner getScanner(String tableName) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            return table.getScanner(scan);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    /**
     * 檢索表中指定數(shù)據(jù)
     *
     * @param tableName  表名
     * @param filterList 過濾器
     */
    public static ResultScanner getScanner(String tableName, FilterList filterList) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            scan.setFilter(filterList);
            return table.getScanner(scan);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    /**
     * 檢索表中指定數(shù)據(jù)
     *
     * @param tableName   表名
     * @param startRowKey 起始 RowKey
     * @param endRowKey   終止 RowKey
     * @param filterList  過濾器
     */
    public static ResultScanner getScanner(String tableName, String startRowKey, String endRowKey,
                                           FilterList filterList) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            scan.setStartRow(Bytes.toBytes(startRowKey));
            scan.setStopRow(Bytes.toBytes(endRowKey));
            scan.setFilter(filterList);
            return table.getScanner(scan);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    /**
     * 刪除指定行記錄
     *
     * @param tableName 表名
     * @param rowKey    唯一標(biāo)識(shí)
     */
    public static boolean deleteRow(String tableName, String rowKey) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return true;
    }
    /**
     * 刪除指定行的指定列
     *
     * @param tableName  表名
     * @param rowKey     唯一標(biāo)識(shí)
     * @param familyName 列族
     * @param qualifier  列標(biāo)識(shí)
     */
    public static boolean deleteColumn(String tableName, String rowKey, String familyName,
                                       String qualifier) {
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            delete.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(qualifier));
            table.delete(delete);
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return true;
    }
}

KafkaProducerUtils Kafka生產(chǎn)者工具類

package com.star.engine.utils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import java.util.Properties;
public class KafkaProducerUtils {
    static Producer<String, String> producer;
    public static void init() {
        Properties props = new Properties();
        //此處配置的是kafka的端口
        props.put("metadata.broker.list", Constants.BROKERS);
        props.put("bootstrap.servers", Constants.BROKERS);
        //配置value的序列化類
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //配置key的序列化類
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("producer.type", "async");
        props.put("request.required.acks", "-1");
        producer = new KafkaProducer<>(props);
    }
    public static Producer getProducer() {
        if (producer == null) {
            init();
        }
        return producer;
    }
}

第一版需求：
     1、完成是否異地登錄判斷

package com.star.engine.core;
import com.alibaba.fastjson.JSON;
import com.star.engine.pojo.ClientLog;
import com.star.engine.utils.Constants;
import com.star.engine.utils.FlinkKafkaUtils;
import com.star.engine.utils.HBaseUtils;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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.util.Collector;
import org.apache.hadoop.hbase.client.HTable;
import org.locationtech.spatial4j.distance.DistanceUtils;
import java.text.SimpleDateFormat;
import java.util.Iterator;
/**
 * 第一版需求：
 *  1、完成是否異地登錄判斷
 *
 */
public class Processl {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        // 1.獲取Kafka數(shù)據(jù)源
        DataStreamSource<String> source = environment.addSource(FlinkKafkaUtils.getKafkaEventSource());
        SingleOutputStreamOperator<ClientLog> clientlogSource = source.map(str -> JSON.parseObject(str, ClientLog.class));
        clientlogSource.keyBy(clientLog -> clientLog.getUserNo())
                .process(new KeyedProcessFunction<String, ClientLog, String>() {
                    /**
                     * 從hbase讀取用戶畫像數(shù)據(jù)
                     * @param parameters
                     * @throws Exception
                     */
                    HTable table;
                    // 基于狀態(tài)存儲(chǔ)用戶上一次登錄的數(shù)據(jù)
                    ListState<ClientLog> privousLoginData;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        table = HBaseUtils.initHbaseClient(Constants.HBASE_TABLE);
                        privousLoginData = getRuntimeContext().getListState(new ListStateDescriptor<ClientLog>("privoulsLoginData", ClientLog.class));
                    }
                    // 對(duì)每條數(shù)據(jù)進(jìn)行處理
                    @Override
                    public void processElement(ClientLog clientLog, Context context, Collector<String> out) throws Exception {
                        String eventType = clientLog.getEventType();
                        if (eventType.equals(Constants.RULE_TYPE_LOGIN)) {
                            // 用戶登錄
                            /**
                             * 1、5分鐘登錄次數(shù)限制
                             * 2、異地登錄
                             * 3、不在常用地區(qū)登錄
                             */
                            // 5分鐘登錄次數(shù)限制
                            // 異地登錄
                            distanceProcess(clientLog,privousLoginData);
                            // 在常用地區(qū)登錄
                        }
                    }
                }).print();
        environment.execute("Processl");
    }
    /**
     * 異地登錄
     * @param clientLog
     * @param privousLoginData
     */
    public static void distanceProcess(ClientLog clientLog, ListState<ClientLog> privousLoginData) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat();
        String score = "0";
        String reason = "第一次登錄：" + clientLog.getAddr() + " 本地登錄時(shí)間：" + clientLog.getDetailTime();
        try {
            // 判斷上一次是否登錄
            Iterator<ClientLog> iterator = privousLoginData.get().iterator();
            if (iterator.hasNext()) {
                ClientLog privousMessage = iterator.next();
                // 不是第一次登錄
                // 靜態(tài)規(guī)則
                String distanceRule = "500|1@400|2@300|3@200|4@100|5";
                String[] distanceRules = distanceRule.split("@");
                String oldTime = privousMessage.getDetailTime();
                String oldGps = privousMessage.getGps();
                String oldAddr = privousMessage.getAddr();
                // 計(jì)算距離
                double distanceReal = DistanceUtils.distHaversineRAD(Double.parseDouble(clientLog.getGps().split("\\,")[0]), Double.parseDouble(clientLog.getGps().split("\\,")[1]), Double.parseDouble(oldGps.split("\\,")[0]), Double.parseDouble(oldGps.split("\\,")[1]));
                // 時(shí)間差
                long time = simpleDateFormat.parse(clientLog.getDetailTime()).getTime() - simpleDateFormat.parse(oldTime).getTime();
                double speed = distanceReal / (time / (1000 * 3600.0));
                // 規(guī)則匹配
                for (String rule : distanceRules) {
                    double speedLimit = Double.parseDouble(rule.split("\\|")[0]);
                    String speedScore = rule.split("\\|")[1];
                    if (speed >= speedLimit) {
                        score = speedScore;
                        reason += "=== 短時(shí)間內(nèi)速度為："+ speed + " 規(guī)定約定速度為：" + speedLimit+" 當(dāng)前登錄地：" + clientLog.getAddr() + " 上一次登錄地：" + privousMessage.getAddr();
                    }
                }
            } else {
                // 第一次登錄
                score = "5";
                reason = "第一次登錄：" + clientLog.getAddr() + " 登錄時(shí)間為：" + clientLog.getDetailTime();
            }
            privousLoginData.clear();
            privousLoginData.add(clientLog);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}