Flink定时器怎么实现
这篇文章主要讲解了“Flink定时器怎么实现”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“Flink定时器怎么实现”吧!
创新互联主要从事成都做网站、成都网站制作、网页设计、企业做网站、公司建网站等业务。立足成都服务徐汇,10年网站建设经验,价格优惠、服务专业,欢迎来电咨询建站服务:028-86922220
背景需求
在电商领域会有这么一个场景,如果用户买了商品,在订单完成之后,24小时之内没有做出评价,系统自动给与五星好评,我们今天主要使用flink的定时器来简单实现这一功能。
案例详解
自定义source
首先我们还是通过自定义source来模拟生成一些订单数据.
在这里,我们生了一个最简单的二元组Tuple2,包含订单id和订单完成时间两个字段.
public static class MySource implements SourceFunction>{
private volatile boolean isRunning = true;
@Override
public void run(SourceContext> ctx) throws Exception{
while (isRunning){
Thread.sleep(1000);
//订单id
String orderid = UUID.randomUUID().toString();
//订单完成时间
long orderFinishTime = System.currentTimeMillis();
ctx.collect(Tuple2.of(orderid, orderFinishTime));
}
}
@Override
public void cancel(){
isRunning = false;
}
}
定时处理逻辑
先上代码, 我们再来依次解释代码
public static class TimerProcessFuntion
extends KeyedProcessFunction,Object>{
private MapState mapState;
//超过多长时间(interval,单位:毫秒) 没有评价,则自动五星好评
private long interval = 0l;
public TimerProcessFuntion(long interval){
this.interval = interval;
}
@Override
public void open(Configuration parameters){
MapStateDescriptor mapStateDesc = new MapStateDescriptor<>(
"mapStateDesc",
String.class, Long.class);
mapState = getRuntimeContext().getMapState(mapStateDesc);
}
@Override
public void onTimer(
long timestamp, OnTimerContext ctx, Collector
首先我们定义一个MapState类型的状态,key是订单号,value是订单完成时间 在processElement处理数据的时候,把每个订单的信息存入状态中,这个时候不做任何处理,并且注册一个比订单完成时间大于间隔时间(interval)的定时器. 注册的定时任务在到达了定时器的时间就会触发onTimer方法,我们主要在这个里面进行处理。我们调用外部的接口来判断用户是否做过评价,如果没做评价,调用接口给与五星好评,如果做过评价,则什么也不处理,最后记得把相应的订单从MapState删除
感谢各位的阅读,以上就是“Flink定时器怎么实现”的内容了,经过本文的学习后,相信大家对Flink定时器怎么实现这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是创新互联,小编将为大家推送更多相关知识点的文章,欢迎关注!
新闻标题:Flink定时器怎么实现
标题来源:http://scgulin.cn/article/jgpgsg.html