阿坝州网站建设公司

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 out) throws Exception{
   Iterator iterator = mapState.iterator();
   while (iterator.hasNext()){
    Map.Entry entry = (Map.Entry) 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 value, Context ctx, Collector out) throws Exception{
   mapState.put(value.f0, value.f1);
   ctx.timerService().registerProcessingTimeTimer(value.f1 + interval);
  }
 }


 
  
  1. 首先我们定义一个MapState类型的状态，key是订单号，value是订单完成时间
  2. 在processElement处理数据的时候，把每个订单的信息存入状态中，这个时候不做任何处理，并且注册一个比订单完成时间大于间隔时间(interval)的定时器.
  3. 注册的定时任务在到达了定时器的时间就会触发onTimer方法，我们主要在这个里面进行处理。我们调用外部的接口来判断用户是否做过评价，如果没做评价，调用接口给与五星好评，如果做过评价，则什么也不处理,最后记得把相应的订单从MapState删除
感谢各位的阅读，以上就是“Flink定时器怎么实现”的内容了，经过本文的学习后，相信大家对Flink定时器怎么实现这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是创新互联，小编将为大家推送更多相关知识点的文章，欢迎关注！
            
            
                            

                当前标题：Flink定时器怎么实现                

                网站地址：http://azwzsj.com/article/jgpgsg.html
            
        
        

            
其他资讯