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

Flink双流实时对账

news 2025/7/16 5:00:51

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

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

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

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

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

©著作权归作者所有,转载或内容合作请联系作者


文章转载自：
http://calcifuge.bdypl.cn
http://cardialgia.bdypl.cn
http://amylopectin.bdypl.cn
http://bbbc.bdypl.cn
http://archicarp.bdypl.cn
http://boldhearted.bdypl.cn
http://atlatl.bdypl.cn
http://astrologous.bdypl.cn
http://agent.bdypl.cn
http://authoress.bdypl.cn
http://chopboat.bdypl.cn
http://chiropractic.bdypl.cn
http://baalish.bdypl.cn
http://athletically.bdypl.cn
http://cavalryman.bdypl.cn
http://busulphan.bdypl.cn
http://bosomy.bdypl.cn
http://aculeate.bdypl.cn
http://chiral.bdypl.cn
http://atacamite.bdypl.cn
http://ameloblast.bdypl.cn
http://appetent.bdypl.cn
http://backflash.bdypl.cn
http://canonic.bdypl.cn
http://cashmere.bdypl.cn
http://appreciably.bdypl.cn
http://bruxelles.bdypl.cn
http://arson.bdypl.cn
http://anopia.bdypl.cn
http://afterbeat.bdypl.cn
http://www.dtcms.com/a/280072.html

相关文章：

  • CPU寄存器、进程上下文与Linux O(1)调度器原理
  • Jfinal+SQLite java工具类复制mysql表数据到 *.sqlite
  • 基于 vue＋Cesium 实现军事标绘之钳击箭头绘制实战
  • C++-linux系统编程 8.进程（三）孤儿进程、僵尸进程与进程回收
  • Oracle学习专栏（五）：性能优化
  • 适用于Windows系统截图工具
  • 通用综合文字识别联动 MES 系统：OCR 是数据流通的核心
  • 【算法-BFS 解决最短路问题】探索BFS在图论中的应用：最短路径问题的高效解法
  • JVM——JVM垃圾回收调优的主要目的是什么？
  • 行为模式-状态模式
  • C++ -- STL-- List
  • 分布式通信框架 - JGroups
  • 从零开始的云计算生活——第三十二天，四面楚歌，HAProxy负载均衡
  • 数据怎么分层？从ODS、DW、ADS三大层一一拆解！
  • 智慧园区：激活城市活力的数字化引擎
  • 【colab 使用uv创建一个新的python版本运行】
  • mac上的app如何自动分类
  • 22-C#的委托简单使用-2
  • 自增主键为什么不是连续的？
  • 基于多智能体强化学习的医疗检索增强生成系统研究—MMOA-RAG架构设计与实现
  • Uboot源码超详细分析（2）
  • 力扣25.7.15每日一题——有效单词
  • 对于编写PID过程中的问题
  • TCP可靠性设计的核心机制与底层逻辑
  • TDengine GREATEST 和 LEAST 函数用户手册
  • 26.将 Python 列表拆分为多个小块
  • SSM框架学习DI入门——day2
  • 跨平台移动开发技术深度分析：uni-app、React Native与Flutter的迁移成本、性能、场景与前景
  • IOS 18下openURL 失效问题
  • Flutter瀑布流布局深度实践：打造高性能动态图片墙