当前位置：首页 > ops >正文

Flink双流实时对账

ops 2025/7/16 10:45:34

在电商、金融、银行、支付等涉及到金钱相关的领域，为了安全起见，一般都有对账的需求。

比如，对于订单支付事件，用户通过某宝付款，虽然用户支付成功，但是用户支付完成后并不算成功，我们得确认平台账户上是否到账了。

针对上述的场景，我们可以采用批处理，或离线计算等技术手段，通过定时任务，每天结束后，扫描数据库中的数据，核对当天的支付数据和交易数据，进行对账。

想要达到实时对账的效果，比如有的用户支付成功但是并没有到账，要及时发出报警，我们必须得依赖实时计算框架。

我们将问题简单化，比如有如下场景，在某电商网站，用户创建订单并支付成功，会将相关信息发给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}

支付成功并且金额已经进入平台账户，往往也会把相关信息发给kafka，如订单id，支付方式、时间等信息。

{orId='order01', payChannel='wechat', timestamp=1606549536}
{orId='order02', payChannel='alipay', timestamp=1606549537}

只有订单在支付(action=pay)成功后，并且成功到账，这才算一次完整的交易。本案例，就是要实时检测那些不成功的交易，如有不成功的，及时发出报警信息。

上述行为本身会产生两种事件流，一种是订单事件流，另一种是交易事件流，我们通过Flink将两种类型的流进行关联，实时分析没有到账的数据，发出报警。

为了简化，我们从socket读取数据流，代替从kafka消费数据。

代码示例

本案例涉及到的知识点：

状态编程
定时器
延迟事件处理
合流操作

首先，我们需要定义订单事件OrderEvents和交易事件ReceiptEvents

// 订单事件
public class OrderEvents {// 用户idprivate Long userId;// 动作private String action;// 订单idprivate String orId;// 时间 单位 sprivate Long timestamp;// get/set
}
// 交易事件
public class ReceiptEvents {// 订单idprivate String orId;// 支付渠道private String payChannel;// 时间 单位 sprivate Long timestamp;// get/set
}

通过Flink程序，联合两条流，实时检测交易失败的数据并输出到侧输出流里。

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") {};// 读取订单数据KeyedStream<OrderEvents, String> orderStream = env.socketTextStream("localhost", 8888).map(new MapFunction<String, OrderEvents>() {@Overridepublic 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>() {@Overridepublic long extractAscendingTimestamp(OrderEvents element) {return element.getTimestamp() * 1000;}}).filter(new FilterFunction<OrderEvents>() {@Overridepublic boolean filter(OrderEvents value) throws Exception {return value.getAction().equals("pay");}}).keyBy(new KeySelector<OrderEvents, String>() {@Overridepublic String getKey(OrderEvents value) throws Exception {return value.getOrId();}});// 读取交易数据KeyedStream<ReceiptEvents, String> receiptStream = env.socketTextStream("localhost", 9999).map(new MapFunction<String, ReceiptEvents>() {@Overridepublic ReceiptEvents map(String value) throws Exception {String[] split = value.split(",");return new ReceiptEvents(split[0], split[1], System.currentTimeMillis() / 1000);}}).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<ReceiptEvents>() {@Overridepublic long extractAscendingTimestamp(ReceiptEvents element) {return element.getTimestamp() * 1000;}}).keyBy(new KeySelector<ReceiptEvents, String>() {@Overridepublic String getKey(ReceiptEvents value) throws Exception {return value.getOrId();}});// connect两条流SingleOutputStreamOperator<String> process = orderStream.connect(receiptStream).process(new MyCoProcessFunction());// 输出正常交易的数据process.print("success");// 输出异常交易的数据process.getSideOutput(payEventTag).print("payEventTag");process.getSideOutput(receiptEventTag).print("receiptEventTag");env.execute("Tx Match job");}

上面代码的主要逻辑是：

从端口为8888和9999的两个socket读取订单事件和交易事件(模拟从kafka消费)，然后将事件数据包装成OrderEvents和ReceiptEvents。
提取事件时间。
对于OrderEvents，只需要action=pay的数据，过滤无用的数据。
将两条流根据orId keyby，生成orderStream和receiptStream，并通过connect合并两条流，将合并后的结果，交给CoProcessFunction函数计算。
将正常交易事件输出在success中，异常的交易事件，输出到两个侧输出流中。

所以，我们需要自定义聚合函数，继承CoProcessFunction函数，实现正常交易和异常交易行为的实时计算。

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") {};// 定义状态,保存订单pay事件和交易事件ValueState<OrderEvents> payEventValueState = null;ValueState<ReceiptEvents> receiptEventValueState = null;@Overridepublic 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事件@Overridepublic 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事件@Overridepublic 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}}// 定时器@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {// 判断哪个状态存在，表示另一个事件没有来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();}
}

上述代码是我们自定义的窗口函数，主要的功能是：

继承了CoProcessFunction，分别在processElement1和procossElement2方法中处理orderEvents和receiptEvent。
定义状态，侧输出流，注册定时器，通过一些逻辑计算是是正常交易还是异常交易。
在processElement1方法中，如果只有pay事件没有receipt事件，则注册一个5s后触发的定时器，等待receipt事件的到来，如果5s后receipt事件仍没有到来，则说明是一个异常交易事件，触发timer并将异常事件输出到侧输出流中。
在processElement2方法中，如果只有receipt事件没有pay事件，表明pay事件和receipt事件乱序，则注册一个3s的定时器，等待pay事件。如果3s后还是没有pay事件到达，则触发timer将延迟的乱序数据输出到侧输出流中。
定义定时器timer，对于异常的交易行为，将交易输出输出到侧输出流。异常交易是指，在一定时间范围内，只有pay事件没有receipt事件或只有receipt事件没有pay事件。如果在一定时间范围内这两个事件都有，则属于正常交易行为。

打开两个socket，输入数据模拟交易行为。为了输出一些异常信息，我们的输入方式，不光要正常输入数据，还要输入一些乱序的数据，比如只输入payEvent不输入receiptEvent等，使之触发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事件，属于异常事件