背景需求

在電商領域會有這么一個場景，如果用戶買了商品，在訂單完成之后，24小時之內沒有做出評價，系統自動給與五星好評，我們今天主要使用flink的定時器來簡單實現這一功能。 

案例詳解 

自定義source

首先我們還是通過自定義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);
  }
 }

1. 首先我們定義一個MapState類型的狀態，key是訂單號，value是訂單完成時間
2. 在processElement處理數據的時候，把每個訂單的信息存入狀態中，這個時候不做任何處理，并且注冊一個比訂單完成時間大于間隔時間(interval)的定時器.
3. 注冊的定時任務在到達了定時器的時間就會觸發onTimer方法，我們主要在這個里面進行處理。我們調用外部的接口來判斷用戶是否做過評價，如果沒做評價，調用接口給與五星好評，如果做過評價，則什么也不處理,最后記得把相應的訂單從MapState刪除