您好,登录后才能下订单哦!
密码登录
登录注册
点击 登录注册 即表示同意《亿速云用户服务条款》
这篇文章主要讲解了“Flink定时器怎么实现”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“Flink定时器怎么实现”吧!
在电商领域会有这么一个场景,如果用户买了商品,在订单完成之后,24小时之内没有做出评价,系统自动给与五星好评,我们今天主要使用flink的定时器来简单实现这一功能。
首先我们还是通过自定义source来模拟生成一些订单数据.
在这里,我们生了一个最简单的二元组Tuple2,包含订单id和订单完成时间两个字段.
public static class MySource implements SourceFunction<Tuple2<String,Long>>{
private volatile boolean isRunning = true;
@Override
public void run(SourceContext<Tuple2<String,Long>> 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<Tuple,Tuple2<String,Long>,Object>{
private MapState<String,Long> mapState;
//超过多长时间(interval,单位:毫秒) 没有评价,则自动五星好评
private long interval = 0l;
public TimerProcessFuntion(long interval){
this.interval = interval;
}
@Override
public void open(Configuration parameters){
MapStateDescriptor<String,Long> mapStateDesc = new MapStateDescriptor<>(
"mapStateDesc",
String.class, Long.class);
mapState = getRuntimeContext().getMapState(mapStateDesc);
}
@Override
public void onTimer(
long timestamp, OnTimerContext ctx, Collector<Object> out) throws Exception{
Iterator iterator = mapState.iterator();
while (iterator.hasNext()){
Map.Entry<String,Long> entry = (Map.Entry<String,Long>) iterator.next();
String orderid = entry.getKey();
boolean f = isEvaluation(entry.getKey());
mapState.remove(orderid);
if (f){
LOG.info("订单(orderid: {}) 在 {} 毫秒时间内已经评价,不做处理", orderid, interval);
}
if (f){
//如果用户没有做评价,在调用相关的接口给与默认的五星评价
LOG.info("订单(orderid: {}) 超过 {} 毫秒未评价,调用接口给与五星自动好评", orderid, interval);
}
}
}
/**
* 用户是否对该订单进行了评价,在生产环境下,可以去查询相关的订单系统.
* 我们这里只是随便做了一个判断
*
* @param key
* @return
*/
private boolean isEvaluation(String key){
return key.hashCode() % 2 == 0;
}
@Override
public void processElement(
Tuple2<String,Long> value, Context ctx, Collector<Object> out) throws Exception{
mapState.put(value.f0, value.f1);
ctx.timerService().registerProcessingTimeTimer(value.f1 + interval);
}
}
感谢各位的阅读,以上就是“Flink定时器怎么实现”的内容了,经过本文的学习后,相信大家对Flink定时器怎么实现这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是亿速云,小编将为大家推送更多相关知识点的文章,欢迎关注!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。