背景
在电商、金融、银行、支付等涉及到金钱相关的领域，为了安全起见，一般都有对账的需求。
比如，对于订单支付事件，用户通过某宝付款，虽然用户支付成功，但是用户支付完成后并不算成功，我们得确认平台账户上是否到账了。
针对上述的场景，我们可以采用批处理，或离线计算等技术手段，通过定时任务，每天结束后，扫描数据库中的数据，核对当天的支付数据和交易数据，进行对账。
想要达到实时对账的效果，比如有的用户支付成功但是并没有到账，要及时发出报警，我们必须得依赖实时计算框架。
我们将问题简单化，比如有如下场景，在某电商网站，用户创建订单并支付成功，会将相关信息发给kafka，字段包括，用户uid、动作、订单id、时间等信息
{userId=1, action='create', orId='order01', timestamp=1606549513}
支付成功并且金额已经进入平台账户，往往也会把相关信息发给kafka，如订单id，支付方式、时间等信息。
{orId='order01', payChannel='wechat', timestamp=1606549536}
只有订单在支付(action=pay)成功后，并且成功到账，这才算一次完整的交易。本案例，就是要实时检测那些不成功的交易，如有不成功的，及时发出报警信息。
上述行为本身会产生两种事件流，一种是订单事件流，另一种是交易事件流，我们通过Flink将两种类型的流进行关联，实时分析没有到账的数据，发出报警。
为了简化，我们从socket读取数据流，代替从kafka消费数据。
代码示例
本案例涉及到的知识点：
状态编程
定时器
延迟事件处理
合流操作
首先，我们需要定义订单事件OrderEvents和交易事件ReceiptEvents
// 订单事件
通过Flink程序，联合两条流，实时检测交易失败的数据并输出到侧输出流里。
public static void main(String[] args) throws Exception {
上面代码的主要逻辑是：
从端口为8888和9999的两个socket读取订单事件和交易事件(模拟从kafka消费)，然后将事件数据包装成OrderEvents和ReceiptEvents。
提取事件时间。
对于OrderEvents，只需要action=pay的数据，过滤无用的数据。
将两条流根据orId keyby，生成orderStream和receiptStream，并通过connect合并两条流，将合并后的结果，交给CoProcessFunction函数计算。
将正常交易事件输出在success中，异常的交易事件，输出到两个侧输出流中。
所以，我们需要自定义聚合函数，继承CoProcessFunction函数，实现正常交易和异常交易行为的实时计算。
class MyCoProcessFunction
上述代码是我们自定义的窗口函数，主要的功能是：
继承了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
输入交易事件
nc -lk 9999
控制台输出：
success> 订单事件：OrderEvents{userId=1, action='pay', orId='orderId01', timestamp=1606555301}和交易事件：ReceiptEvents{orId='orderId01', payEquipment='wechat', timestamp=1606555307} 属于正常交易
好了今天的分享就到这，谢谢大家的阅读，如有疑问请留言或扫描下面二维码加笔者微信，一起交流。
扫描二维码
加笔者微信
获得更多精彩
本文分享自微信公众号 - OutOfMemoryError（backend_technology）。
如有侵权，请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。
