flink实战-定时器实现已完成订单自动五星好评

Z时代
2024-01-10
分类：技术分享

背景需求

在电商领域会有这么一个场景，如果用户买了商品，在订单完成之后，24小时之内没有做出评价，系统自动给与五星好评，我们今天主要使用flink的定时器来简单实现这一功能。

案例详解

自定义source

首先我们还是通过自定义source来模拟生成一些订单数据.

在这里，我们生了一个最简单的二元组Tuple2,包含订单id和订单完成时间两个字段.


publicstaticclassMySourceimplementsSourceFunction<Tuple2<String,Long>>{
privatevolatileboolean isRunning = true;
@Override
publicvoidrun(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
publicvoidcancel(){
   isRunning = false;
  }
 }

定时处理逻辑

先上代码, 我们再来依次解释代码


publicstaticclassTimerProcessFuntion
extendsKeyedProcessFunction<Tuple,Tuple2<String,Long>,Object>{
private MapState<String,Long> mapState;
//超过多长时间(interval,单位：毫秒) 没有评价，则自动五星好评
privatelong interval = 0l;
publicTimerProcessFuntion(long interval){
this.interval = interval;
  }
@Override
publicvoidopen(Configuration parameters){
   MapStateDescriptor<String,Long> mapStateDesc = new MapStateDescriptor<>(
"mapStateDesc",
     String.class, Long.class);
   mapState = getRuntimeContext().getMapState(mapStateDesc);
  }
@Override
publicvoidonTimer(
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
   */
privatebooleanisEvaluation(String key){
return key.hashCode() % 2 == 0;
  }
@Override
publicvoidprocessElement(
    Tuple2<String,Long> value, Context ctx, Collector<Object> out)throws Exception{
   mapState.put(value.f0, value.f1);
   ctx.timerService().registerProcessingTimeTimer(value.f1 + interval);
  }
 }