背景
在電商、金融闰围、銀行赃绊、支付等涉及到金錢相關(guān)的領(lǐng)域既峡，為了安全起見羡榴，一般都有對賬的需求。
比如运敢，對于訂單支付事件校仑，用戶通過某寶付款，雖然用戶支付成功传惠，但是用戶支付完成后并不算成功迄沫，我們得確認(rèn)平臺賬戶上是否到賬了。
針對上述的場景卦方，我們可以采用批處理羊瘩，或離線計算等技術(shù)手段，通過定時任務(wù)盼砍，每天結(jié)束后尘吗，掃描數(shù)據(jù)庫中的數(shù)據(jù)，核對當(dāng)天的支付數(shù)據(jù)和交易數(shù)據(jù)浇坐，進行對賬睬捶。
想要達到實時對賬的效果，比如有的用戶支付成功但是并沒有到賬近刘，要及時發(fā)出報警擒贸，我們必須得依賴實時計算框架。
我們將問題簡單化觉渴，比如有如下場景介劫，在某電商網(wǎng)站，用戶創(chuàng)建訂單并支付成功案淋，會將相關(guān)信息發(fā)給kafka座韵，字段包括，用戶uid哎迄、動作回右、訂單id隆圆、時間等信息
{userId=1, action='create', orId='order01', timestamp=1606549513}
{userId=1, action='pay', orId='order01', timestamp=1606549516}
{userId=2, action='create', orId='order02', timestamp=1606549513}
支付成功并且金額已經(jīng)進入平臺賬戶，往往也會把相關(guān)信息發(fā)給kafka翔烁，如訂單id渺氧，支付方式、時間等信息蹬屹。
{orId='order01', payChannel='wechat', timestamp=1606549536}
{orId='order02', payChannel='alipay', timestamp=1606549537}
只有訂單在支付(action=pay)成功后侣背，并且成功到賬，這才算一次完整的交易慨默。本案例贩耐，就是要實時檢測那些不成功的交易，如有不成功的厦取，及時發(fā)出報警信息潮太。
上述行為本身會產(chǎn)生兩種事件流，一種是訂單事件流虾攻，另一種是交易事件流铡买，我們通過Flink將兩種類型的流進行關(guān)聯(lián)，實時分析沒有到賬的數(shù)據(jù)霎箍，發(fā)出報警奇钞。
為了簡化，我們從socket讀取數(shù)據(jù)流漂坏，代替從kafka消費數(shù)據(jù)景埃。
代碼示例
本案例涉及到的知識點：
- 狀態(tài)編程
- 定時器
- 延遲事件處理
- 合流操作
首先，我們需要定義訂單事件OrderEvents和交易事件ReceiptEvents
// 訂單事件
public class OrderEvents {
// 用戶id
private Long userId;
// 動作
private String action;
// 訂單id
private String orId;
// 時間 單位 s
private Long timestamp;
// get/set
}
// 交易事件
public class ReceiptEvents {
// 訂單id
private String orId;
// 支付渠道
private String payChannel;
// 時間 單位 s
private Long timestamp;
// get/set
}
通過Flink程序顶别，聯(lián)合兩條流谷徙，實時檢測交易失敗的數(shù)據(jù)并輸出到側(cè)輸出流里。
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
// 定義測輸出流筋夏，輸出只有pay事件沒有receipt事件的異常信息
OutputTag payEventTag = new OutputTag<String>("payEventTag-side") {};
// 定義測輸出流蒂胞，輸出只有receipt事件沒有pay事件的異常信息
OutputTag receiptEventTag = new OutputTag<String>("receiptEventTag-side") {};
// 讀取訂單數(shù)據(jù)
KeyedStream<OrderEvents, String> orderStream = env.socketTextStream("localhost", 8888).map(new MapFunction<String, OrderEvents>() {
@Override
public OrderEvents map(String value) throws Exception {
String[] split = value.split(",");
return new OrderEvents(Long.parseLong(split[0]), split[1], split[2], System.currentTimeMillis() / 1000);
}
}).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvents>() {
@Override
public long extractAscendingTimestamp(OrderEvents element) {
return element.getTimestamp() * 1000;
}
}).filter(new FilterFunction<OrderEvents>() {
@Override
public boolean filter(OrderEvents value) throws Exception {
return value.getAction().equals("pay");
}
}).keyBy(new KeySelector<OrderEvents, String>() {
@Override
public String getKey(OrderEvents value) throws Exception {
return value.getOrId();
}
});
// 讀取交易數(shù)據(jù)
KeyedStream<ReceiptEvents, String> receiptStream = env.socketTextStream("localhost", 9999).map(new MapFunction<String, ReceiptEvents>() {
@Override
public ReceiptEvents map(String value) throws Exception {
String[] split = value.split(",");
return new ReceiptEvents(split[0], split[1], System.currentTimeMillis() / 1000);
}
}).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<ReceiptEvents>() {
@Override
public long extractAscendingTimestamp(ReceiptEvents element) {
return element.getTimestamp() * 1000;
}
}).keyBy(new KeySelector<ReceiptEvents, String>() {
@Override
public String getKey(ReceiptEvents value) throws Exception {
return value.getOrId();
}
});
// connect兩條流
SingleOutputStreamOperator<String> process = orderStream.connect(receiptStream).process(new MyCoProcessFunction());
// 輸出正常交易的數(shù)據(jù)
process.print("success");
// 輸出異常交易的數(shù)據(jù)
process.getSideOutput(payEventTag).print("payEventTag");
process.getSideOutput(receiptEventTag).print("receiptEventTag");
env.execute("Tx Match job");
}
上面代碼的主要邏輯是：
從端口為8888和9999的兩個socket讀取訂單事件和交易事件(模擬從kafka消費)，然后將事件數(shù)據(jù)包裝成OrderEvents和ReceiptEvents条篷。
提取事件時間骗随。
對于OrderEvents，只需要action=pay的數(shù)據(jù)赴叹，過濾無用的數(shù)據(jù)鸿染。
將兩條流根據(jù)orId keyby，生成orderStream和receiptStream乞巧，并通過connect合并兩條流涨椒，將合并后的結(jié)果，交給CoProcessFunction函數(shù)計算。
將正常交易事件輸出在success中蚕冬，異常的交易事件免猾，輸出到兩個側(cè)輸出流中。
所以囤热，我們需要自定義聚合函數(shù)猎提，繼承CoProcessFunction函數(shù)，實現(xiàn)正常交易和異常交易行為的實時計算旁蔼。
class MyCoProcessFunction
extends CoProcessFunction<OrderEvents, ReceiptEvents, String> {
// 定義測輸出流锨苏，輸出只有pay事件沒有receipt事件的異常信息
OutputTag payEventTag = new OutputTag<String>("payEventTag-side") {};
// 定義測輸出流，輸出只有receipt事件沒有pay事件的異常信息
OutputTag receiptEventTag = new OutputTag<String>("receiptEventTag-side") {};
// 定義狀態(tài),保存訂單pay事件和交易事件
ValueState<OrderEvents> payEventValueState = null;
ValueState<ReceiptEvents> receiptEventValueState = null;
@Override
public void open(Configuration parameters) throws Exception {
ValueStateDescriptor<OrderEvents> descriptor1
= new ValueStateDescriptor<OrderEvents>("payEventValueState", OrderEvents.class);
ValueStateDescriptor<ReceiptEvents> descriptor2
= new ValueStateDescriptor<ReceiptEvents>("receiptEventValueState", ReceiptEvents.class);
payEventValueState = getRuntimeContext().getState(descriptor1);
receiptEventValueState = getRuntimeContext().getState(descriptor2);
}
// 處理OrderEvents事件
@Override
public void processElement1(OrderEvents orderEvents, Context ctx, Collector<String> out) throws Exception {
if (receiptEventValueState.value() != null) {
// 正常輸出匹配
out.collect("訂單事件："+orderEvents.toString() + "和交易事件：" + receiptEventValueState.value().toString());
receiptEventValueState.clear();
payEventValueState.clear();
} else {
// 如果沒有到賬事件棺聊，注冊定時器等待
payEventValueState.update(orderEvents);
ctx.timerService().registerEventTimeTimer(orderEvents.getTimestamp() * 1000 + 5000L); // 5s
}
}
// 處理receipt事件
@Override
public void processElement2(ReceiptEvents receiptEvents, Context ctx, Collector<String> out) throws Exception {
if (payEventValueState.value() != null) {
// 正常輸出
out.collect("訂單事件："+payEventValueState.value().toString() + "和交易事件：" + receiptEvents.toString()+" 屬于正常交易");
receiptEventValueState.clear();
payEventValueState.clear();
} else {
// 如果沒有訂單事件伞租，說明是亂序事件，注冊定時器等待
receiptEventValueState.update(receiptEvents);
ctx.timerService().registerEventTimeTimer(receiptEvents.getTimestamp() * 1000 + 3000L); // 3s
}
}
// 定時器
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
// 判斷哪個狀態(tài)存在限佩，表示另一個事件沒有來
if (payEventValueState.value() != null) {
ctx.output(payEventTag, payEventValueState.value().toString() + " 有pay事件沒有receipt事件葵诈，屬于異常事件");
}
if (receiptEventValueState.value() != null) {
ctx.output(receiptEventTag, receiptEventValueState.value().toString() + " 有receipt事件沒有pay事件。屬于異常事件");
}
receiptEventValueState.clear();
payEventValueState.clear();
}
}
上述代碼是我們自定義的窗口函數(shù)犀暑，主要的功能是：
繼承了CoProcessFunction驯击，分別在processElement1和procossElement2方法中處理orderEvents和receiptEvent烁兰。
定義狀態(tài)耐亏，側(cè)輸出流，注冊定時器沪斟，通過一些邏輯計算是是正常交易還是異常交易广辰。
在processElement1方法中，如果只有pay事件沒有receipt事件主之，則注冊一個5s后觸發(fā)的定時器择吊，等待receipt事件的到來，如果5s后receipt事件仍沒有到來槽奕，則說明是一個異常交易事件几睛，觸發(fā)timer并將異常事件輸出到側(cè)輸出流中。
在processElement2方法中粤攒，如果只有receipt事件沒有pay事件所森，表明pay事件和receipt事件亂序，則注冊一個3s的定時器夯接，等待pay事件焕济。如果3s后還是沒有pay事件到達，則觸發(fā)timer將延遲的亂序數(shù)據(jù)輸出到側(cè)輸出流中盔几。
定義定時器timer晴弃，對于異常的交易行為，將交易輸出輸出到側(cè)輸出流。異常交易是指上鞠，在一定時間范圍內(nèi)际邻，只有pay事件沒有receipt事件 或 只有receipt事件沒有pay事件。如果在一定時間范圍內(nèi)這兩個事件都有芍阎，則屬于正常交易行為枯怖。
打開兩個socket，輸入數(shù)據(jù)模擬交易行為能曾。為了輸出一些異常信息度硝，我們的輸入方式，不光要正常輸入數(shù)據(jù)寿冕，還要輸入一些亂序的數(shù)據(jù)蕊程，比如只輸入payEvent不輸入receiptEvent等，使之觸發(fā)timer驼唱。
輸入訂單事件
nc -lk 8888
1,pay,orderId01
2,pay,orderId02
3,pay,orderId03
4,pay,orderId04
6,pay,orderId06
7,pay,orderId07
8,pay,orderId0
輸入交易事件
nc -lk 9999
orderId01,wechat
orderId03,alipay
orderId04,wechat
orderId05,alipay
orderId06,wechat
orderId08,alipa
控制臺輸出：
success> 訂單事件：OrderEvents{userId=1, action='pay', orId='orderId01', timestamp=1606555301}和交易事件：ReceiptEvents{orId='orderId01', payEquipment='wechat', timestamp=1606555307} 屬于正常交易
success> 訂單事件：OrderEvents{userId=3, action='pay', orId='orderId03', timestamp=1606555318}和交易事件：ReceiptEvents{orId='orderId03', payEquipment='alipay', timestamp=1606555325} 屬于正常交易
payEventTag> OrderEvents{userId=2, action='pay', orId='orderId02', timestamp=1606555313} 有pay事件沒有receipt事件藻茂，屬于異常事件
success> 訂單事件：OrderEvents{userId=4, action='pay', orId='orderId04', timestamp=1606555332}和交易事件：ReceiptEvents{orId='orderId04', payEquipment='wechat', timestamp=1606555338} 屬于正常交易
success> 訂單事件：OrderEvents{userId=6, action='pay', orId='orderId06', timestamp=1606555351}和交易事件：ReceiptEvents{orId='orderId06', payEquipment='wechat', timestamp=1606555358} 屬于正常交易
receiptEventTag> ReceiptEvents{orId='orderId05', payEquipment='alipay', timestamp=1606555345} 有receipt事件沒有pay事件。屬于異常事件
success> 訂單事件：OrderEvents{userId=8, action='pay', orId='orderId08', timestamp=1606555375}和交易事件：ReceiptEvents{orId='orderId08', payEquipment='alipay', timestamp=1606555382} 屬于正常交易
payEventTag> OrderEvents{userId=7, action='pay', orId='orderId07', timestamp=1606555368} 有pay事件沒有receipt事件玫恳，屬于異常事件
