1、思路分析
在电商网站中，订单的支付作为直接与营销收入挂钩的一环，在业务流程中非常重要。对于订单而言，为了正确控制业务流程，也为了增加用户的支付意愿，网站一般会设置一个支付失效时间，超过一段时间不支付的订单就会被取消。
在电商平台中，最终创造收入和利润的是用户下单购买的环节；更具体一点，是用户真正完成支付动作的时候。用户下单的行为可以表明用户对商品的需求，但在现实中，并不是每次下单都会被用户立刻支付。当拖延一段时间后，用户支付的意愿会降低。所以为了让用户更有紧迫感从而提高支付转化率，同时也为了防范订单支付环节的安全风险，电商网站往往会对订单状态进行监控，设置一个失效时间（比如15分钟），如果下单后一段时间仍未支付，订单就会被取消。
我们同样可以利用Process Function，自定义实现检测订单超时的功能。为了简化问题，我们只考虑超时报警的情形，在pay事件超时未发生的情况下，输出超时报警信息。
一个简单的思路是，可以在订单的create事件到来后注册定时器，15分钟后触发；然后再用一个布尔类型的Value状态来作为标识位，表明pay事件是否发生过。如果pay事件已经发生，状态被置为true，那么就不再需要做什么操作；而如果pay事件一直没来，状态一直为false，到定时器触发时，就应该输出超时报警信息。
2、具体实现
我们准备了订单数据作为分析数据源OrderLog.csv，这组数据中有订单创建的相关信息，也有订单支付的相关信息。
具体实现的代码如下：
public class OrderTimeoutWithoutCep {
// 定义超时事件的侧输出流标签
private final static OutputTag<OrderResult> orderTimeoutTag = new OutputTag<OrderResult>("order-timeout"){};
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.setParallelism(1);
// 读取数据并转换成POJO类型
URL resource = OrderPayTimeout.class.getResource("/OrderLog.csv");
DataStream<OrderEvent> orderEventStream = env.readTextFile(resource.getPath())
.map(line -> {
String[] fields = line.split(",");
return new OrderEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
})
.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
@Override
public long extractAscendingTimestamp(OrderEvent element) {
return element.getTimestamp() * 1000L;
}
});
// 自定义处理函数，主流输出正常匹配订单事件，侧输出流输出超时报警事件
SingleOutputStreamOperator<OrderResult> resultStream = orderEventStream
.keyBy(OrderEvent::getOrderId)
.process(new OrderPayMatchDetect());
resultStream.print("payed normally");
resultStream.getSideOutput(orderTimeoutTag).print("timeout");
env.execute("order timeout detect without cep job");
}
// 实现自定义KeyedProcessFunction
public static class OrderPayMatchDetect extends KeyedProcessFunction<Long, OrderEvent, OrderResult>{
// 定义状态，保存之前点单是否已经来过create、pay的事件
ValueState<Boolean> isPayedState;
ValueState<Boolean> isCreatedState;
// 定义状态，保存定时器时间戳
ValueState<Long> timerTsState;
@Override
public void open(Configuration parameters) throws Exception {
isPayedState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("is-payed", Boolean.class, false));
isCreatedState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("is-created", Boolean.class, false));
timerTsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-ts", Long.class));
}
@Override
public void processElement(OrderEvent value, Context ctx, Collector<OrderResult> out) throws Exception {
// 先获取当前状态
Boolean isPayed = isPayedState.value();
Boolean isCreated = isCreatedState.value();
Long timerTs = timerTsState.value();
// 判断当前事件类型
if( "create".equals(value.getEventType()) ){
// 1. 如果来的是create，要判断是否支付过
if( isPayed ){
// 1.1 如果已经正常支付，输出正常匹配结果
out.collect(new OrderResult(value.getOrderId(), "payed successfully"));
// 清空状态，删除定时器
isCreatedState.clear();
isPayedState.clear();
timerTsState.clear();
ctx.timerService().deleteEventTimeTimer(timerTs);
} else {
// 1.2 如果没有支付过，注册15分钟后的定时器，开始等待支付事件
Long ts = ( value.getTimestamp() + 15 * 60 ) * 1000L;
ctx.timerService().registerEventTimeTimer(ts);
// 更新状态
timerTsState.update(ts);
isCreatedState.update(true);
}
} else if( "pay".equals(value.getEventType()) ){
// 2. 如果来的是pay，要判断是否有下单事件来过
if( isCreated ){
// 2.1 已经有过下单事件，要继续判断支付的时间戳是否超过15分钟
if( value.getTimestamp() * 1000L < timerTs ){
// 2.1.1 在15分钟内，没有超时，正常匹配输出
out.collect(new OrderResult(value.getOrderId(), "payed successfully"));
} else {
// 2.1.2 已经超时，输出侧输出流报警
ctx.output(orderTimeoutTag, new OrderResult(value.getOrderId(), "payed but already timeout"));
}
// 统一清空状态
isCreatedState.clear();
isPayedState.clear();
timerTsState.clear();
ctx.timerService().deleteEventTimeTimer(timerTs);
} else {
// 2.2 没有下单事件，乱序，注册一个定时器，等待下单事件
ctx.timerService().registerEventTimeTimer( value.getTimestamp() * 1000L);
// 更新状态
timerTsState.update(value.getTimestamp() * 1000L);
isPayedState.update(true);
}
}
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<OrderResult> out) throws Exception {
// 定时器触发，说明一定有一个事件没来
if( isPayedState.value() ){
// 如果pay来了，说明create没来
ctx.output(orderTimeoutTag, new OrderResult(ctx.getCurrentKey(), "payed but not found created log"));
}else {
// 如果pay没来，支付超时
ctx.output(orderTimeoutTag, new OrderResult(ctx.getCurrentKey(), "timeout"));
}
// 清空状态
isCreatedState.clear();
isPayedState.clear();
timerTsState.clear();
}
}
}
想要了解跟多关于
课程内容欢迎关注尚硅谷大数据培训，尚硅谷除了这些技术文章外还有免费的高质量大数据培训课程视频供广大学员下载学习。
上一篇: Standalone模式运行机制_大数据培训
下一篇: 推荐算法！基于隐语义模型的协同过滤推荐之用户商品推荐列表
